reimport this PoC

7 files changed, 1244 insertions, 0 deletions

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.cl