From 3fd46f3977fb650974e5e936cba362c787c00637 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 7 Feb 2026 16:32:10 +0200 Subject: reimport this PoC --- internal/ingester/auto.go | 145 +++++++++++ internal/ingester/auto_test.go | 164 ++++++++++++ internal/ingester/clickhouse.go | 191 ++++++++++++++ internal/ingester/pushgateway.go | 51 ++++ internal/ingester/pushgateway_test.go | 28 +++ internal/ingester/remotewrite.go | 455 ++++++++++++++++++++++++++++++++++ internal/ingester/remotewrite_test.go | 210 ++++++++++++++++ 7 files changed, 1244 insertions(+) create mode 100644 internal/ingester/auto.go create mode 100644 internal/ingester/auto_test.go create mode 100644 internal/ingester/clickhouse.go create mode 100644 internal/ingester/pushgateway.go create mode 100644 internal/ingester/pushgateway_test.go create mode 100644 internal/ingester/remotewrite.go create mode 100644 internal/ingester/remotewrite_test.go (limited to 'internal/ingester') diff --git a/internal/ingester/auto.go b/internal/ingester/auto.go new file mode 100644 index 0000000..315d767 --- /dev/null +++ b/internal/ingester/auto.go @@ -0,0 +1,145 @@ +package ingester + +import ( + "context" + "fmt" + "log" + "time" + + "epimetheus/internal/config" + "epimetheus/internal/metrics" +) + +const ageThreshold = 5 * time.Minute + +// DetermineMode automatically determines which ingestion mode to use based on timestamp age. +// Data older than 5 minutes uses historic mode (Remote Write), newer data uses realtime mode (Pushgateway). +func DetermineMode(timestamp time.Time) config.Mode { + age := time.Since(timestamp) + if age > ageThreshold { + return config.ModeHistoric + } + return config.ModeRealtime +} + +// AutoIngester handles automatic ingestion by routing samples to appropriate ingesters. +type AutoIngester struct { + pushgateway PushgatewayIngester + remoteWrite RemoteWriteIngester + collectors metrics.Collectors +} + +// NewAutoIngester creates a new auto ingester. +func NewAutoIngester(collectors metrics.Collectors) AutoIngester { + return AutoIngester{ + pushgateway: NewPushgatewayIngester(), + remoteWrite: NewRemoteWriteIngester(), + collectors: collectors, + } +} + +// Ingest automatically routes samples to appropriate ingestion method based on timestamp age. +func (a AutoIngester) Ingest(ctx context.Context, samples []metrics.Sample, cfg config.Config) error { + if len(samples) == 0 { + return fmt.Errorf("no samples to ingest") + } + + realtimeSamples, historicSamples := groupSamplesByMode(samples) + + logIngestSummary(len(samples), len(realtimeSamples), len(historicSamples)) + + if len(realtimeSamples) > 0 { + if err := a.ingestRealtime(ctx, cfg); err != nil { + return fmt.Errorf("failed to ingest realtime samples: %w", err) + } + } + + if len(historicSamples) > 0 { + if err := a.ingestHistoric(ctx, historicSamples, cfg); err != nil { + return fmt.Errorf("failed to ingest historic samples: %w", err) + } + } + + log.Printf("\nšŸŽ‰ Auto-ingest complete!") + return nil +} + +// groupSamplesByMode separates samples into realtime and historic groups. +func groupSamplesByMode(samples []metrics.Sample) (realtime, historic []metrics.Sample) { + realtimeSamples := make([]metrics.Sample, 0) + historicSamples := make([]metrics.Sample, 0) + + for _, sample := range samples { + if DetermineMode(sample.Timestamp) == config.ModeRealtime { + realtimeSamples = append(realtimeSamples, sample) + } else { + historicSamples = append(historicSamples, sample) + } + } + + return realtimeSamples, historicSamples +} + +// logIngestSummary logs the ingestion summary. +func logIngestSummary(total, realtime, historic int) { + log.Printf("šŸ“Š Auto-ingest summary:") + log.Printf(" Total samples: %d", total) + log.Printf(" Realtime samples (< 5min old): %d", realtime) + log.Printf(" Historic samples (> 5min old): %d", historic) +} + +// ingestRealtime ingests realtime samples via Pushgateway. +func (a AutoIngester) ingestRealtime(ctx context.Context, cfg config.Config) error { + log.Printf("\nšŸ”„ Ingesting REALTIME samples via Pushgateway...") + log.Printf(" Note: Pushgateway uses current timestamp (original timestamps ignored)") + + if err := a.pushgateway.Ingest(ctx, a.collectors, cfg.PushgatewayURL, cfg.JobName); err != nil { + return err + } + + log.Printf("āœ… Successfully ingested realtime samples") + return nil +} + +// ingestHistoric ingests historic samples via Remote Write. +func (a AutoIngester) ingestHistoric(ctx context.Context, samples []metrics.Sample, cfg config.Config) error { + log.Printf("\nā° Ingesting %d HISTORIC samples via Remote Write...", len(samples)) + + // Log a few sample details instead of all samples + samplesToLog := 3 + if len(samples) < samplesToLog { + samplesToLog = len(samples) + } + + for i := 0; i < samplesToLog; i++ { + age := time.Since(samples[i].Timestamp) + log.Printf(" Sample %d: %s (age: %s)", i+1, samples[i].MetricName, formatDuration(age)) + } + + if len(samples) > samplesToLog { + // Show oldest and newest sample ages + oldestAge := time.Since(samples[0].Timestamp) + newestAge := time.Since(samples[len(samples)-1].Timestamp) + log.Printf(" ... (%d more samples)", len(samples)-samplesToLog) + log.Printf(" Age range: %s (oldest) to %s (newest)", formatDuration(oldestAge), formatDuration(newestAge)) + } + + if err := a.remoteWrite.Ingest(ctx, samples, cfg.PrometheusURL); err != nil { + return err + } + + log.Printf("āœ… Successfully ingested %d historic samples", len(samples)) + return nil +} + +// formatDuration formats a duration in human-readable form. +func formatDuration(d time.Duration) string { + if d < time.Minute { + return fmt.Sprintf("%.0f seconds", d.Seconds()) + } else if d < time.Hour { + return fmt.Sprintf("%.0f minutes", d.Minutes()) + } else if d < 24*time.Hour { + return fmt.Sprintf("%.1f hours", d.Hours()) + } + return fmt.Sprintf("%.1f days", d.Hours()/24) +} diff --git a/internal/ingester/auto_test.go b/internal/ingester/auto_test.go new file mode 100644 index 0000000..fc1c423 --- /dev/null +++ b/internal/ingester/auto_test.go @@ -0,0 +1,164 @@ +package ingester + +import ( + "context" + "testing" + "time" + + "epimetheus/internal/config" + "epimetheus/internal/metrics" +) + +func TestDetermineMode(t *testing.T) { + tests := []struct { + name string + timestamp time.Time + want config.Mode + }{ + { + name: "current time is realtime", + timestamp: time.Now(), + want: config.ModeRealtime, + }, + { + name: "1 minute ago is realtime", + timestamp: time.Now().Add(-1 * time.Minute), + want: config.ModeRealtime, + }, + { + name: "4 minutes ago is realtime", + timestamp: time.Now().Add(-4 * time.Minute), + want: config.ModeRealtime, + }, + { + name: "6 minutes ago is historic", + timestamp: time.Now().Add(-6 * time.Minute), + want: config.ModeHistoric, + }, + { + name: "1 hour ago is historic", + timestamp: time.Now().Add(-1 * time.Hour), + want: config.ModeHistoric, + }, + { + name: "1 day ago is historic", + timestamp: time.Now().Add(-24 * time.Hour), + want: config.ModeHistoric, + }, + { + name: "exactly 5 minutes is historic (edge case)", + timestamp: time.Now().Add(-5 * time.Minute), + want: config.ModeHistoric, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := DetermineMode(tt.timestamp) + if got != tt.want { + age := time.Since(tt.timestamp) + t.Errorf("DetermineMode() = %v, want %v (age: %v)", got, tt.want, age) + } + }) + } +} + +func TestGroupSamplesByMode(t *testing.T) { + now := time.Now() + samples := []metrics.Sample{ + {MetricName: "metric1", Timestamp: now.Add(-1 * time.Minute)}, // realtime + {MetricName: "metric2", Timestamp: now.Add(-2 * time.Minute)}, // realtime + {MetricName: "metric3", Timestamp: now.Add(-10 * time.Minute)}, // historic + {MetricName: "metric4", Timestamp: now.Add(-1 * time.Hour)}, // historic + {MetricName: "metric5", Timestamp: now.Add(-30 * time.Second)}, // realtime + } + + realtime, historic := groupSamplesByMode(samples) + + if len(realtime) != 3 { + t.Errorf("Got %d realtime samples, want 3", len(realtime)) + } + if len(historic) != 2 { + t.Errorf("Got %d historic samples, want 2", len(historic)) + } + + // Verify correct grouping + for _, s := range realtime { + if DetermineMode(s.Timestamp) != config.ModeRealtime { + t.Errorf("Sample %s incorrectly grouped as realtime (age: %v)", s.MetricName, s.Age()) + } + } + for _, s := range historic { + if DetermineMode(s.Timestamp) != config.ModeHistoric { + t.Errorf("Sample %s incorrectly grouped as historic (age: %v)", s.MetricName, s.Age()) + } + } +} + +func TestFormatDuration(t *testing.T) { + tests := []struct { + name string + duration time.Duration + want string + }{ + { + name: "seconds", + duration: 45 * time.Second, + want: "45 seconds", + }, + { + name: "minutes", + duration: 5 * time.Minute, + want: "5 minutes", + }, + { + name: "hours", + duration: 2*time.Hour + 30*time.Minute, + want: "2.5 hours", + }, + { + name: "days", + duration: 36 * time.Hour, + want: "1.5 days", + }, + { + name: "less than minute", + duration: 30 * time.Second, + want: "30 seconds", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := formatDuration(tt.duration) + if got != tt.want { + t.Errorf("formatDuration() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestAutoIngester_Ingest_EmptySamples(t *testing.T) { + collectors := metrics.NewCollectors() + autoIngester := NewAutoIngester(collectors) + ctx := context.Background() + cfg := config.NewConfig() + + err := autoIngester.Ingest(ctx, []metrics.Sample{}, cfg) + if err == nil { + t.Error("Expected error for empty samples, got nil") + } + if err.Error() != "no samples to ingest" { + t.Errorf("Expected 'no samples to ingest' error, got: %v", err) + } +} + +func TestAutoIngester_New(t *testing.T) { + collectors := metrics.NewCollectors() + ingester := NewAutoIngester(collectors) + + // Verify ingester was created with components + if ingester.collectors.RequestsTotal == nil { + t.Error("AutoIngester.collectors not initialized properly") + } +} diff --git a/internal/ingester/clickhouse.go b/internal/ingester/clickhouse.go new file mode 100644 index 0000000..2439d4e --- /dev/null +++ b/internal/ingester/clickhouse.go @@ -0,0 +1,191 @@ +package ingester + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "net/url" + "strings" + "sync" + "time" + + "epimetheus/internal/metrics" +) + +const ( + clickhouseBatchSize = 10000 + clickhouseHTTPTimeout = 30 * time.Second + defaultTableName = "epimetheus_metrics" +) + +// ClickHouseIngester ingests metrics into ClickHouse. +type ClickHouseIngester struct { + client *http.Client + baseURL string + tableName string +} + +// NewClickHouseIngester creates a new ClickHouse ingester. +func NewClickHouseIngester(baseURL, tableName string) ClickHouseIngester { + if tableName == "" { + tableName = defaultTableName + } + // Ensure URL has scheme and no trailing slash for query params + baseURL = strings.TrimSuffix(strings.TrimSpace(baseURL), "/") + + return ClickHouseIngester{ + client: &http.Client{ + Timeout: clickhouseHTTPTimeout, + }, + baseURL: baseURL, + tableName: tableName, + } +} + +// EnsureTable creates the metrics table if it does not exist. +func (c ClickHouseIngester) EnsureTable(ctx context.Context) error { + createSQL := fmt.Sprintf(` + CREATE TABLE IF NOT EXISTS %s ( + metric String, + labels Map(String, String), + value Float64, + timestamp DateTime64(3) + ) ENGINE = MergeTree() + ORDER BY (metric, timestamp) + `, c.tableName) + + reqURL := c.baseURL + "/?query=" + url.QueryEscape(createSQL) + req, err := http.NewRequestWithContext(ctx, "POST", reqURL, nil) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("execute create table: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("create table failed (status %d): %s", resp.StatusCode, string(body)) + } + + return nil +} + +// Ingest inserts samples into ClickHouse in batches. +func (c ClickHouseIngester) Ingest(ctx context.Context, samples []metrics.Sample) error { + if len(samples) == 0 { + return fmt.Errorf("no samples to ingest") + } + + if err := c.EnsureTable(ctx); err != nil { + return fmt.Errorf("ensure table: %w", err) + } + + batches := chunkSamplesForClickHouse(samples, clickhouseBatchSize) + totalBatches := len(batches) + + log.Printf("ClickHouse: ingesting %d samples in %d batches", len(samples), totalBatches) + + var wg sync.WaitGroup + errChan := make(chan error, totalBatches) + + for i, batch := range batches { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + wg.Add(1) + go func(idx int, b []metrics.Sample) { + defer wg.Done() + if err := c.insertBatch(ctx, b); err != nil { + errChan <- fmt.Errorf("batch %d: %w", idx+1, err) + } + }(i, batch) + } + + wg.Wait() + close(errChan) + + var firstErr error + for err := range errChan { + if firstErr == nil { + firstErr = err + } + log.Printf("ClickHouse batch error: %v", err) + } + + return firstErr +} + +// insertBatch sends a single batch via HTTP JSONEachRow. +func (c ClickHouseIngester) insertBatch(ctx context.Context, samples []metrics.Sample) error { + var buf bytes.Buffer + enc := json.NewEncoder(&buf) + + for _, s := range samples { + // Convert labels map to format ClickHouse Map expects + labelsMap := make(map[string]string, len(s.Labels)) + for k, v := range s.Labels { + labelsMap[k] = v + } + + row := struct { + Metric string `json:"metric"` + Labels map[string]string `json:"labels"` + Value float64 `json:"value"` + Timestamp string `json:"timestamp"` + }{ + Metric: s.MetricName, + Labels: labelsMap, + Value: s.Value, + Timestamp: s.Timestamp.UTC().Format("2006-01-02 15:04:05.000"), + } + + if err := enc.Encode(row); err != nil { + return fmt.Errorf("encode row: %w", err) + } + } + + query := fmt.Sprintf("INSERT INTO %s FORMAT JSONEachRow", c.tableName) + reqURL := c.baseURL + "/?query=" + url.QueryEscape(query) + + req, err := http.NewRequestWithContext(ctx, "POST", reqURL, &buf) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("send request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("insert failed (status %d): %s", resp.StatusCode, string(body)) + } + + return nil +} + +func chunkSamplesForClickHouse(samples []metrics.Sample, size int) [][]metrics.Sample { + var batches [][]metrics.Sample + for i := 0; i < len(samples); i += size { + end := i + size + if end > len(samples) { + end = len(samples) + } + batches = append(batches, samples[i:end]) + } + return batches +} diff --git a/internal/ingester/pushgateway.go b/internal/ingester/pushgateway.go new file mode 100644 index 0000000..c5c80c3 --- /dev/null +++ b/internal/ingester/pushgateway.go @@ -0,0 +1,51 @@ +package ingester + +import ( + "context" + "fmt" + + "epimetheus/internal/metrics" + + "github.com/prometheus/client_golang/prometheus/push" +) + +// PushgatewayIngester handles realtime metric ingestion via Pushgateway. +// Note: Pushgateway does not preserve custom timestamps - all metrics are +// timestamped with the current time when pushed. +type PushgatewayIngester struct{} + +// NewPushgatewayIngester creates a new Pushgateway ingester. +func NewPushgatewayIngester() PushgatewayIngester { + return PushgatewayIngester{} +} + +// Ingest pushes metrics to Pushgateway. +// The samples parameter is currently ignored because Pushgateway doesn't support +// custom metric values from samples - it uses registered Prometheus collectors. +// This ingests generated metrics using the provided collectors. +func (i PushgatewayIngester) Ingest(ctx context.Context, collectors metrics.Collectors, url, jobName string) error { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + // Generate random metric values + collectors.Simulate() + + // Create pusher with all collectors + pusher := push.New(url, jobName). + Collector(collectors.RequestsTotal). + Collector(collectors.ActiveConnections). + Collector(collectors.TemperatureCelsius). + Collector(collectors.RequestDuration). + Collector(collectors.JobsProcessed). + Grouping("instance", "example-app") + + // Push metrics to Pushgateway + if err := pusher.Push(); err != nil { + return fmt.Errorf("failed to push to pushgateway: %w", err) + } + + return nil +} diff --git a/internal/ingester/pushgateway_test.go b/internal/ingester/pushgateway_test.go new file mode 100644 index 0000000..55fc213 --- /dev/null +++ b/internal/ingester/pushgateway_test.go @@ -0,0 +1,28 @@ +package ingester + +import ( + "testing" + + "epimetheus/internal/metrics" +) + +func TestNewPushgatewayIngester(t *testing.T) { + ingester := NewPushgatewayIngester() + + // Verify the ingester was created (value type, so no nil check needed) + _ = ingester +} + +func TestPushgatewayIngester_Type(t *testing.T) { + // Test that we can create and use the ingester + collectors := metrics.NewCollectors() + ingester := NewPushgatewayIngester() + + // The ingester should work with collectors + if collectors.RequestsTotal == nil { + t.Error("Collectors not initialized properly") + } + + // Verify ingester is the correct type + _ = ingester +} diff --git a/internal/ingester/remotewrite.go b/internal/ingester/remotewrite.go new file mode 100644 index 0000000..b88b7fa --- /dev/null +++ b/internal/ingester/remotewrite.go @@ -0,0 +1,455 @@ +package ingester + +import ( + "bytes" + "context" + "fmt" + "io" + "log" + "math/rand" + "net/http" + "sync" + "sync/atomic" + "time" + + "epimetheus/internal/metrics" + + "github.com/golang/snappy" + "github.com/prometheus/prometheus/prompb" +) + +const ( + requestTimeout = 10 * time.Second + backfillDelay = 100 * time.Millisecond + // BatchSize defines the maximum number of samples per Remote Write request + // Prometheus has a 32MB limit, so we keep batches small to avoid rejection + BatchSize = 5000 + // NumWorkers defines the number of concurrent goroutines for batch processing + // Higher values increase throughput but may overwhelm Prometheus + NumWorkers = 10 +) + +// Buffer pool for reusing buffers across requests (reduces GC pressure) +var bufferPool = sync.Pool{ + New: func() interface{} { + return new(bytes.Buffer) + }, +} + +// Preallocated buffer pool for protobuf marshaling +var protoBufferPool = sync.Pool{ + New: func() interface{} { + // Preallocate ~500KB buffer (typical batch size) + buf := make([]byte, 0, 512*1024) + return &buf + }, +} + +// TimeSeries object pool for reuse +var timeSeriesPool = sync.Pool{ + New: func() interface{} { + return &prompb.TimeSeries{ + Labels: make([]prompb.Label, 0, 10), // Typical label count + Samples: make([]prompb.Sample, 0, 1), + } + }, +} + +// Label slice pool for reuse +var labelSlicePool = sync.Pool{ + New: func() interface{} { + labels := make([]prompb.Label, 0, 10) + return &labels + }, +} + +// RemoteWriteIngester handles historic metric ingestion via Prometheus Remote Write API. +// This ingester preserves custom timestamps, making it suitable for importing historic data. +type RemoteWriteIngester struct { + client *http.Client +} + +// NewRemoteWriteIngester creates a new Remote Write ingester with optimized HTTP client. +func NewRemoteWriteIngester() RemoteWriteIngester { + // Optimized HTTP transport with connection pooling + transport := &http.Transport{ + MaxIdleConns: 100, // Global connection pool + MaxIdleConnsPerHost: NumWorkers, // Match worker count + MaxConnsPerHost: NumWorkers, // Limit concurrent connections per host + IdleConnTimeout: 90 * time.Second, // Keep connections alive longer + DisableKeepAlives: false, // CRITICAL: enable keep-alive + DisableCompression: true, // We compress manually with Snappy + ForceAttemptHTTP2: true, // Use HTTP/2 if available + WriteBufferSize: 64 * 1024, // 64KB write buffer + ReadBufferSize: 64 * 1024, // 64KB read buffer + } + + return RemoteWriteIngester{ + client: &http.Client{ + Timeout: requestTimeout, + Transport: transport, + }, + } +} + +// Ingest sends samples to Prometheus via Remote Write API with batching and concurrency. +// Large datasets are automatically split into batches and processed by a worker pool +// to avoid exceeding Prometheus's 32MB limit and maximize throughput. +func (i RemoteWriteIngester) Ingest(ctx context.Context, samples []metrics.Sample, url string) error { + if len(samples) == 0 { + return fmt.Errorf("no samples to ingest") + } + + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + // Split samples into batches + batches := chunkSamples(samples, BatchSize) + totalBatches := len(batches) + + log.Printf("Splitting %d samples into %d batches (batch size: %d)", len(samples), totalBatches, BatchSize) + log.Printf("Processing batches with %d concurrent workers", NumWorkers) + + // Counters for tracking progress (atomic for thread safety) + var successCount int32 + var errorCount int32 + var processedCount int32 + + // Worker pool pattern + batchChan := make(chan batchJob, totalBatches) + errorsChan := make(chan error, totalBatches) + var wg sync.WaitGroup + + // Start workers + for w := 0; w < NumWorkers; w++ { + wg.Add(1) + go func(workerID int) { + defer wg.Done() + + for job := range batchChan { + // Check context before processing + select { + case <-ctx.Done(): + errorsChan <- fmt.Errorf("worker %d cancelled", workerID) + return + default: + } + + // Convert and send batch + timeSeries := convertSamplesToTimeSeries(job.batch) + writeRequest := &prompb.WriteRequest{Timeseries: timeSeries} + + if err := i.sendWriteRequest(ctx, url, writeRequest); err != nil { + atomic.AddInt32(&errorCount, 1) + errorsChan <- fmt.Errorf("batch %d: %w", job.index, err) + } else { + atomic.AddInt32(&successCount, 1) + } + + // Update progress + processed := atomic.AddInt32(&processedCount, 1) + if processed%10 == 0 || int(processed) == totalBatches { + progress := float64(processed) / float64(totalBatches) * 100 + log.Printf("Progress: %.1f%% (%d/%d batches, %d success, %d errors)", + progress, processed, totalBatches, atomic.LoadInt32(&successCount), atomic.LoadInt32(&errorCount)) + } + } + }(w) + } + + // Send batches to workers + for idx, batch := range batches { + batchChan <- batchJob{index: idx + 1, batch: batch} + } + close(batchChan) + + // Wait for all workers to finish + wg.Wait() + close(errorsChan) + + // Collect errors + var firstError error + errorList := make([]error, 0) + for err := range errorsChan { + errorList = append(errorList, err) + if firstError == nil { + firstError = err + } + } + + finalSuccess := int(atomic.LoadInt32(&successCount)) + finalErrors := int(atomic.LoadInt32(&errorCount)) + + log.Printf("Batch ingestion complete: %d successful, %d errors", finalSuccess, finalErrors) + + if finalErrors > 0 { + // Log first few errors as examples + numToLog := 5 + if len(errorList) < numToLog { + numToLog = len(errorList) + } + log.Printf("Sample errors (showing %d of %d):", numToLog, finalErrors) + for i := 0; i < numToLog; i++ { + log.Printf(" - %v", errorList[i]) + } + return fmt.Errorf("completed with %d/%d batches failed", finalErrors, totalBatches) + } + + return nil +} + +// batchJob represents a batch to be processed by a worker. +type batchJob struct { + index int + batch []metrics.Sample +} + +// chunkSamples splits samples into batches of the specified size. +func chunkSamples(samples []metrics.Sample, batchSize int) [][]metrics.Sample { + var batches [][]metrics.Sample + + for i := 0; i < len(samples); i += batchSize { + end := i + batchSize + if end > len(samples) { + end = len(samples) + } + batches = append(batches, samples[i:end]) + } + + return batches +} + +// IngestHistoric generates and ingests historic metrics for a specific time in the past. +func (i RemoteWriteIngester) IngestHistoric(ctx context.Context, url string, hoursAgo int) error { + timestamp := time.Now().Add(-time.Duration(hoursAgo) * time.Hour) + timeSeries := generateHistoricTimeSeries(timestamp) + writeRequest := &prompb.WriteRequest{Timeseries: timeSeries} + + if err := i.sendWriteRequest(ctx, url, writeRequest); err != nil { + return err + } + + log.Printf("Successfully pushed historic data for %d hours ago (timestamp: %s)", + hoursAgo, timestamp.Format(time.RFC3339)) + return nil +} + +// Backfill ingests historic metrics for a range of time points. +func (i RemoteWriteIngester) Backfill(ctx context.Context, url string, startHoursAgo, endHoursAgo, intervalHours int) error { + log.Printf("Starting backfill from %d hours ago to %d hours ago (interval: %d hours)", + startHoursAgo, endHoursAgo, intervalHours) + + successCount := 0 + errorCount := 0 + + for hoursAgo := startHoursAgo; hoursAgo >= endHoursAgo; hoursAgo -= intervalHours { + if err := i.IngestHistoric(ctx, url, hoursAgo); err != nil { + log.Printf("Error pushing data for %d hours ago: %v", hoursAgo, err) + errorCount++ + } else { + successCount++ + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backfillDelay): + } + } + + log.Printf("Backfill complete: %d successful, %d errors", successCount, errorCount) + + if errorCount > 0 { + return fmt.Errorf("backfill completed with %d errors", errorCount) + } + + return nil +} + +// sendWriteRequest sends a write request to Prometheus using pooled buffers. +func (i RemoteWriteIngester) sendWriteRequest(ctx context.Context, url string, writeRequest *prompb.WriteRequest) error { + // Get protobuf buffer from pool + protoBufPtr := protoBufferPool.Get().(*[]byte) + protoBuf := (*protoBufPtr)[:0] // Reset length but keep capacity + defer protoBufferPool.Put(protoBufPtr) + + // Marshal into pooled buffer + data, err := writeRequest.Marshal() + if err != nil { + return fmt.Errorf("failed to marshal write request: %w", err) + } + + // Compress using pooled buffer + compressed := snappy.Encode(protoBuf, data) + + // Get request buffer from pool + buf := bufferPool.Get().(*bytes.Buffer) + buf.Reset() + defer bufferPool.Put(buf) + + buf.Write(compressed) + + req, err := http.NewRequestWithContext(ctx, "POST", url, buf) + if err != nil { + return fmt.Errorf("failed to create HTTP request: %w", err) + } + + req.Header.Set("Content-Type", "application/x-protobuf") + req.Header.Set("Content-Encoding", "snappy") + req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0") + req.Header.Set("Content-Length", fmt.Sprintf("%d", buf.Len())) + + resp, err := i.client.Do(req) + if err != nil { + return fmt.Errorf("failed to send remote write request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("remote write failed with status %d: %s", resp.StatusCode, string(body)) + } + + return nil +} + +// convertSamplesToTimeSeries converts metrics.Sample to prompb.TimeSeries format with pooling. +func convertSamplesToTimeSeries(samples []metrics.Sample) []prompb.TimeSeries { + // Preallocate with exact capacity to avoid reallocation + timeSeries := make([]prompb.TimeSeries, 0, len(samples)) + + // Reusable label slice + labelsPtr := labelSlicePool.Get().(*[]prompb.Label) + labels := *labelsPtr + defer labelSlicePool.Put(labelsPtr) + + for i := range samples { + sample := &samples[i] // Avoid copying + + // Reset labels slice for reuse + labels = labels[:0] + + // Add __name__ label + labels = append(labels, prompb.Label{Name: "__name__", Value: sample.MetricName}) + + // Add custom labels + for k, v := range sample.Labels { + labels = append(labels, prompb.Label{Name: k, Value: v}) + } + + // Copy labels (must not share slice across time series) + labelsCopy := make([]prompb.Label, len(labels)) + copy(labelsCopy, labels) + + // Create time series (reuse pattern, but we need unique objects) + timeSeries = append(timeSeries, prompb.TimeSeries{ + Labels: labelsCopy, + Samples: []prompb.Sample{{ + Value: sample.Value, + Timestamp: sample.Timestamp.UnixMilli(), + }}, + }) + } + + return timeSeries +} + +// generateHistoricTimeSeries generates example time series for a specific timestamp. +func generateHistoricTimeSeries(timestamp time.Time) []prompb.TimeSeries { + timestampMs := timestamp.UnixMilli() + var timeSeries []prompb.TimeSeries + + baseLabels := []prompb.Label{ + {Name: "instance", Value: "example-app"}, + {Name: "job", Value: "historic_data"}, + } + + timeSeries = append(timeSeries, createCounterSeries("epimetheus_test_requests_total", baseLabels, float64(rand.Intn(100)+1), timestampMs)) + timeSeries = append(timeSeries, createGaugeSeries("epimetheus_test_active_connections", baseLabels, float64(rand.Intn(100)), timestampMs)) + timeSeries = append(timeSeries, createGaugeSeries("epimetheus_test_temperature_celsius", baseLabels, 15+rand.Float64()*20, timestampMs)) + + timeSeries = append(timeSeries, generateHistogramSeries(baseLabels, timestampMs)...) + timeSeries = append(timeSeries, generateLabeledCounterSeries(baseLabels, timestampMs)...) + + return timeSeries +} + +// createCounterSeries creates a counter metric time series. +func createCounterSeries(name string, baseLabels []prompb.Label, value float64, timestamp int64) prompb.TimeSeries { + labels := []prompb.Label{{Name: "__name__", Value: name}} + labels = append(labels, baseLabels...) + + return prompb.TimeSeries{ + Labels: labels, + Samples: []prompb.Sample{{Value: value, Timestamp: timestamp}}, + } +} + +// createGaugeSeries creates a gauge metric time series. +func createGaugeSeries(name string, baseLabels []prompb.Label, value float64, timestamp int64) prompb.TimeSeries { + return createCounterSeries(name, baseLabels, value, timestamp) +} + +// generateHistogramSeries generates histogram bucket time series. +func generateHistogramSeries(baseLabels []prompb.Label, timestamp int64) []prompb.TimeSeries { + buckets := []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10} + var series []prompb.TimeSeries + + cumulativeCount := 0 + for _, bucket := range buckets { + cumulativeCount += rand.Intn(5) + labels := []prompb.Label{ + {Name: "__name__", Value: "epimetheus_test_request_duration_seconds_bucket"}, + {Name: "le", Value: fmt.Sprintf("%g", bucket)}, + } + labels = append(labels, baseLabels...) + + series = append(series, prompb.TimeSeries{ + Labels: labels, + Samples: []prompb.Sample{{Value: float64(cumulativeCount), Timestamp: timestamp}}, + }) + } + + infLabels := []prompb.Label{ + {Name: "__name__", Value: "epimetheus_test_request_duration_seconds_bucket"}, + {Name: "le", Value: "+Inf"}, + } + infLabels = append(infLabels, baseLabels...) + series = append(series, prompb.TimeSeries{ + Labels: infLabels, + Samples: []prompb.Sample{{Value: float64(cumulativeCount), Timestamp: timestamp}}, + }) + + series = append(series, createCounterSeries("epimetheus_test_request_duration_seconds_sum", baseLabels, rand.Float64()*100, timestamp)) + series = append(series, createCounterSeries("epimetheus_test_request_duration_seconds_count", baseLabels, float64(cumulativeCount), timestamp)) + + return series +} + +// generateLabeledCounterSeries generates labeled counter time series. +func generateLabeledCounterSeries(baseLabels []prompb.Label, timestamp int64) []prompb.TimeSeries { + jobTypes := []string{"email", "report", "backup"} + statuses := []string{"success", "failed"} + var series []prompb.TimeSeries + + for _, jobType := range jobTypes { + for _, status := range statuses { + labels := []prompb.Label{ + {Name: "__name__", Value: "epimetheus_test_jobs_processed_total"}, + {Name: "job_type", Value: jobType}, + {Name: "status", Value: status}, + } + labels = append(labels, baseLabels...) + + series = append(series, prompb.TimeSeries{ + Labels: labels, + Samples: []prompb.Sample{{Value: float64(rand.Intn(20)), Timestamp: timestamp}}, + }) + } + } + + return series +} diff --git a/internal/ingester/remotewrite_test.go b/internal/ingester/remotewrite_test.go new file mode 100644 index 0000000..d1fdee6 --- /dev/null +++ b/internal/ingester/remotewrite_test.go @@ -0,0 +1,210 @@ +package ingester + +import ( + "testing" + "time" + + "epimetheus/internal/metrics" + + "github.com/prometheus/prometheus/prompb" +) + +func TestNewRemoteWriteIngester(t *testing.T) { + ingester := NewRemoteWriteIngester() + if ingester.client == nil { + t.Error("RemoteWriteIngester.client should not be nil") + } +} + +func TestConvertSamplesToTimeSeries(t *testing.T) { + now := time.Now() + samples := []metrics.Sample{ + { + MetricName: "test_metric1", + Labels: map[string]string{"env": "prod", "host": "server1"}, + Value: 42.5, + Timestamp: now, + }, + { + MetricName: "test_metric2", + Labels: map[string]string{"env": "test"}, + Value: 100.0, + Timestamp: now.Add(-1 * time.Hour), + }, + } + + timeSeries := convertSamplesToTimeSeries(samples) + + if len(timeSeries) != 2 { + t.Errorf("Expected 2 time series, got %d", len(timeSeries)) + } + + // Check first time series + ts1 := timeSeries[0] + if len(ts1.Labels) != 3 { // __name__ + 2 custom labels + t.Errorf("Expected 3 labels, got %d", len(ts1.Labels)) + } + + hasName := false + for _, label := range ts1.Labels { + if label.Name == "__name__" && label.Value == "test_metric1" { + hasName = true + } + } + if !hasName { + t.Error("Missing or incorrect __name__ label") + } + + if len(ts1.Samples) != 1 { + t.Errorf("Expected 1 sample, got %d", len(ts1.Samples)) + } + if ts1.Samples[0].Value != 42.5 { + t.Errorf("Expected value 42.5, got %f", ts1.Samples[0].Value) + } +} + +func TestGenerateHistoricTimeSeries(t *testing.T) { + timestamp := time.Now().Add(-24 * time.Hour) + + timeSeries := generateHistoricTimeSeries(timestamp) + + if len(timeSeries) == 0 { + t.Error("Expected time series to be generated") + } + + // Should contain various metric types + metricNames := make(map[string]bool) + for _, ts := range timeSeries { + for _, label := range ts.Labels { + if label.Name == "__name__" { + metricNames[label.Value] = true + } + } + } + + expectedMetrics := []string{ + "epimetheus_test_requests_total", + "epimetheus_test_active_connections", + "epimetheus_test_temperature_celsius", + "epimetheus_test_jobs_processed_total", + } + + for _, expected := range expectedMetrics { + if !metricNames[expected] { + t.Errorf("Expected metric %s not found", expected) + } + } +} + +func TestCreateCounterSeries(t *testing.T) { + baseLabels := []prompb.Label{ + {Name: "instance", Value: "test-instance"}, + {Name: "job", Value: "test-job"}, + } + + ts := createCounterSeries("test_counter", baseLabels, 123.45, 1234567890000) + + if len(ts.Labels) != 3 { // __name__ + 2 base labels + t.Errorf("Expected 3 labels, got %d", len(ts.Labels)) + } + + if len(ts.Samples) != 1 { + t.Errorf("Expected 1 sample, got %d", len(ts.Samples)) + } + + if ts.Samples[0].Value != 123.45 { + t.Errorf("Expected value 123.45, got %f", ts.Samples[0].Value) + } + + if ts.Samples[0].Timestamp != 1234567890000 { + t.Errorf("Expected timestamp 1234567890000, got %d", ts.Samples[0].Timestamp) + } +} + +func TestCreateGaugeSeries(t *testing.T) { + baseLabels := []prompb.Label{ + {Name: "instance", Value: "test-instance"}, + } + + ts := createGaugeSeries("test_gauge", baseLabels, 67.89, 9876543210000) + + if len(ts.Samples) != 1 { + t.Errorf("Expected 1 sample, got %d", len(ts.Samples)) + } + + if ts.Samples[0].Value != 67.89 { + t.Errorf("Expected value 67.89, got %f", ts.Samples[0].Value) + } +} + +func TestGenerateHistogramSeries(t *testing.T) { + baseLabels := []prompb.Label{ + {Name: "instance", Value: "test-instance"}, + } + timestamp := int64(1234567890000) + + series := generateHistogramSeries(baseLabels, timestamp) + + if len(series) == 0 { + t.Error("Expected histogram series to be generated") + } + + // Should contain buckets, +Inf, sum, and count + metricTypes := make(map[string]int) + for _, ts := range series { + for _, label := range ts.Labels { + if label.Name == "__name__" { + metricTypes[label.Value]++ + } + } + } + + if metricTypes["epimetheus_test_request_duration_seconds_bucket"] == 0 { + t.Error("Expected histogram buckets") + } + if metricTypes["epimetheus_test_request_duration_seconds_sum"] != 1 { + t.Error("Expected histogram sum") + } + if metricTypes["epimetheus_test_request_duration_seconds_count"] != 1 { + t.Error("Expected histogram count") + } +} + +func TestGenerateLabeledCounterSeries(t *testing.T) { + baseLabels := []prompb.Label{ + {Name: "instance", Value: "test-instance"}, + } + timestamp := int64(1234567890000) + + series := generateLabeledCounterSeries(baseLabels, timestamp) + + if len(series) == 0 { + t.Error("Expected labeled counter series to be generated") + } + + // Should have combinations of job types and statuses + // 3 job types * 2 statuses = 6 series + if len(series) != 6 { + t.Errorf("Expected 6 labeled counter series, got %d", len(series)) + } + + // Verify label structure + for _, ts := range series { + hasJobType := false + hasStatus := false + for _, label := range ts.Labels { + if label.Name == "job_type" { + hasJobType = true + } + if label.Name == "status" { + hasStatus = true + } + } + if !hasJobType { + t.Error("Expected job_type label") + } + if !hasStatus { + t.Error("Expected status label") + } + } +} -- cgit v1.2.3