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/config/config.go | 57 ++++ internal/config/config_test.go | 52 ++++ 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 +++++++++++++++ internal/metrics/generator.go | 85 ++++++ internal/metrics/generator_test.go | 53 ++++ internal/metrics/sample.go | 34 +++ internal/metrics/sample_test.go | 160 +++++++++++ internal/parser/csv.go | 101 +++++++ internal/parser/csv_test.go | 175 ++++++++++++ internal/parser/json.go | 62 +++++ internal/parser/json_test.go | 177 +++++++++++++ internal/parser/parser.go | 56 ++++ internal/parser/parser_test.go | 99 +++++++ internal/parser/tabular_csv.go | 256 ++++++++++++++++++ internal/parser/tabular_csv_test.go | 469 +++++++++++++++++++++++++++++++++ internal/resolver/dns_resolver.go | 274 +++++++++++++++++++ internal/resolver/dns_resolver_test.go | 232 ++++++++++++++++ internal/version/version.go | 4 + internal/watcher/file_watcher.go | 86 ++++++ 25 files changed, 3676 insertions(+) create mode 100644 internal/config/config.go create mode 100644 internal/config/config_test.go 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 create mode 100644 internal/metrics/generator.go create mode 100644 internal/metrics/generator_test.go create mode 100644 internal/metrics/sample.go create mode 100644 internal/metrics/sample_test.go create mode 100644 internal/parser/csv.go create mode 100644 internal/parser/csv_test.go create mode 100644 internal/parser/json.go create mode 100644 internal/parser/json_test.go create mode 100644 internal/parser/parser.go create mode 100644 internal/parser/parser_test.go create mode 100644 internal/parser/tabular_csv.go create mode 100644 internal/parser/tabular_csv_test.go create mode 100644 internal/resolver/dns_resolver.go create mode 100644 internal/resolver/dns_resolver_test.go create mode 100644 internal/version/version.go create mode 100644 internal/watcher/file_watcher.go (limited to 'internal') diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..dea804b --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,57 @@ +package config + +import "time" + +// Mode represents the ingestion mode +type Mode string + +const ( + ModeRealtime Mode = "realtime" + ModeHistoric Mode = "historic" + ModeBackfill Mode = "backfill" + ModeAuto Mode = "auto" + ModeWatch Mode = "watch" +) + +// Config holds all configuration for the prometheus-pusher +type Config struct { + Mode Mode + PushgatewayURL string + PrometheusURL string + ClickHouseURL string // ClickHouse HTTP URL (e.g. http://localhost:8123) + ClickHouseTable string // ClickHouse table name (default: epimetheus_metrics) + JobName string + Continuous bool + InputFile string + InputFormat string + MetricName string + HoursAgo int + StartHours int + EndHours int + Interval int + ResolveIPLabels []string // Labels containing IP addresses to resolve via DNS +} + +// NewConfig creates a new Config with default values +func NewConfig() Config { + return Config{ + Mode: ModeRealtime, + PushgatewayURL: "http://localhost:9091", + PrometheusURL: "http://localhost:9090/api/v1/write", + ClickHouseURL: "", + ClickHouseTable: "epimetheus_metrics", + JobName: "example_metrics_pusher", + InputFormat: "csv", + HoursAgo: 24, + StartHours: 48, + EndHours: 0, + Interval: 1, + ResolveIPLabels: []string{"ip"}, // Default resolves 'ip' label + } +} + +// AutoIngestThreshold is the age threshold for auto mode routing +const AutoIngestThreshold = 5 * time.Minute + +// DefaultHTTPTimeout is the default timeout for HTTP requests +const DefaultHTTPTimeout = 10 * time.Second diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..da073c4 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,52 @@ +package config + +import ( + "testing" + "time" +) + +func TestNewConfig(t *testing.T) { + cfg := NewConfig() + + if cfg.Mode != ModeRealtime { + t.Errorf("Default mode = %v, want %v", cfg.Mode, ModeRealtime) + } + if cfg.PushgatewayURL != "http://localhost:9091" { + t.Errorf("Default PushgatewayURL = %v, want http://localhost:9091", cfg.PushgatewayURL) + } + if cfg.PrometheusURL != "http://localhost:9090/api/v1/write" { + t.Errorf("Default PrometheusURL = %v, want http://localhost:9090/api/v1/write", cfg.PrometheusURL) + } + if cfg.JobName != "example_metrics_pusher" { + t.Errorf("Default JobName = %v, want example_metrics_pusher", cfg.JobName) + } + if cfg.InputFormat != "csv" { + t.Errorf("Default InputFormat = %v, want csv", cfg.InputFormat) + } + if cfg.HoursAgo != 24 { + t.Errorf("Default HoursAgo = %v, want 24", cfg.HoursAgo) + } + if cfg.Interval != 1 { + t.Errorf("Default Interval = %v, want 1", cfg.Interval) + } +} + +func TestModeConstants(t *testing.T) { + modes := []Mode{ModeRealtime, ModeHistoric, ModeBackfill, ModeAuto} + expected := []string{"realtime", "historic", "backfill", "auto"} + + for i, mode := range modes { + if string(mode) != expected[i] { + t.Errorf("Mode constant %d = %v, want %v", i, mode, expected[i]) + } + } +} + +func TestConstants(t *testing.T) { + if AutoIngestThreshold != 5*time.Minute { + t.Errorf("AutoIngestThreshold = %v, want 5m", AutoIngestThreshold) + } + if DefaultHTTPTimeout != 10*time.Second { + t.Errorf("DefaultHTTPTimeout = %v, want 10s", DefaultHTTPTimeout) + } +} 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") + } + } +} diff --git a/internal/metrics/generator.go b/internal/metrics/generator.go new file mode 100644 index 0000000..c85906a --- /dev/null +++ b/internal/metrics/generator.go @@ -0,0 +1,85 @@ +package metrics + +import ( + "math/rand" + + "github.com/prometheus/client_golang/prometheus" +) + +const ( + minTemperature = 15.0 + maxTemperature = 35.0 + maxConnections = 100 + maxRequests = 10 +) + +var ( + jobTypes = []string{"email", "report", "backup"} + statuses = []string{"success", "failed"} +) + +// Collectors holds Prometheus metric collectors for realtime mode +type Collectors struct { + RequestsTotal prometheus.Counter + ActiveConnections prometheus.Gauge + TemperatureCelsius prometheus.Gauge + RequestDuration prometheus.Histogram + JobsProcessed *prometheus.CounterVec +} + +// NewCollectors creates new Prometheus metric collectors for testing. +// All metrics are prefixed with "epimetheus_test_" to clearly indicate +// they are generated by the prometheus-pusher test/demo functionality. +func NewCollectors() Collectors { + return Collectors{ + RequestsTotal: prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "epimetheus_test_requests_total", + Help: "Total number of requests processed (test metric)", + }, + ), + ActiveConnections: prometheus.NewGauge( + prometheus.GaugeOpts{ + Name: "epimetheus_test_active_connections", + Help: "Number of currently active connections (test metric)", + }, + ), + TemperatureCelsius: prometheus.NewGauge( + prometheus.GaugeOpts{ + Name: "epimetheus_test_temperature_celsius", + Help: "Current temperature in Celsius (test metric)", + }, + ), + RequestDuration: prometheus.NewHistogram( + prometheus.HistogramOpts{ + Name: "epimetheus_test_request_duration_seconds", + Help: "Histogram of request duration in seconds (test metric)", + Buckets: prometheus.DefBuckets, + }, + ), + JobsProcessed: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "epimetheus_test_jobs_processed_total", + Help: "Total number of jobs processed by type (test metric)", + }, + []string{"job_type", "status"}, + ), + } +} + +// Simulate generates random metric values for the collectors +func (c Collectors) Simulate() { + c.RequestsTotal.Add(float64(rand.Intn(maxRequests) + 1)) + c.ActiveConnections.Set(float64(rand.Intn(maxConnections))) + c.TemperatureCelsius.Set(minTemperature + rand.Float64()*(maxTemperature-minTemperature)) + + for i := 0; i < rand.Intn(5)+1; i++ { + duration := rand.Float64() * 2 + c.RequestDuration.Observe(duration) + } + + for _, jobType := range jobTypes { + status := statuses[rand.Intn(len(statuses))] + c.JobsProcessed.WithLabelValues(jobType, status).Add(float64(rand.Intn(5))) + } +} diff --git a/internal/metrics/generator_test.go b/internal/metrics/generator_test.go new file mode 100644 index 0000000..69395eb --- /dev/null +++ b/internal/metrics/generator_test.go @@ -0,0 +1,53 @@ +package metrics + +import ( + "testing" +) + +func TestNewCollectors(t *testing.T) { + collectors := NewCollectors() + + if collectors.RequestsTotal == nil { + t.Error("RequestsTotal should not be nil") + } + if collectors.ActiveConnections == nil { + t.Error("ActiveConnections should not be nil") + } + if collectors.TemperatureCelsius == nil { + t.Error("TemperatureCelsius should not be nil") + } + if collectors.RequestDuration == nil { + t.Error("RequestDuration should not be nil") + } + if collectors.JobsProcessed == nil { + t.Error("JobsProcessed should not be nil") + } +} + +func TestCollectors_Simulate(t *testing.T) { + collectors := NewCollectors() + + // Should not panic + collectors.Simulate() + + // Run multiple times to test randomness + for i := 0; i < 10; i++ { + collectors.Simulate() + } +} + +func TestCollectors_SimulateMetrics(t *testing.T) { + collectors := NewCollectors() + + // Test that metrics get values after simulation + collectors.Simulate() + + // We can't easily inspect the values without the prometheus client, + // but we can verify the collectors were created properly + if collectors.RequestsTotal == nil { + t.Error("RequestsTotal not initialized") + } + if collectors.JobsProcessed == nil { + t.Error("JobsProcessed not initialized") + } +} diff --git a/internal/metrics/sample.go b/internal/metrics/sample.go new file mode 100644 index 0000000..04360f5 --- /dev/null +++ b/internal/metrics/sample.go @@ -0,0 +1,34 @@ +package metrics + +import "time" + +// Sample represents a single metric sample with timestamp +type Sample struct { + MetricName string + Labels map[string]string + Value float64 + Timestamp time.Time +} + +// NewSample creates a new Sample +func NewSample(name string, labels map[string]string, value float64, timestamp time.Time) Sample { + if labels == nil { + labels = make(map[string]string) + } + return Sample{ + MetricName: name, + Labels: labels, + Value: value, + Timestamp: timestamp, + } +} + +// Age returns how old the sample is +func (s Sample) Age() time.Duration { + return time.Since(s.Timestamp) +} + +// IsRecent returns true if the sample is recent enough for realtime ingestion +func (s Sample) IsRecent(threshold time.Duration) bool { + return s.Age() < threshold +} diff --git a/internal/metrics/sample_test.go b/internal/metrics/sample_test.go new file mode 100644 index 0000000..2ffd78b --- /dev/null +++ b/internal/metrics/sample_test.go @@ -0,0 +1,160 @@ +package metrics + +import ( + "testing" + "time" +) + +func TestNewSample(t *testing.T) { + tests := []struct { + name string + metric string + labels map[string]string + value float64 + timestamp time.Time + wantNil bool + }{ + { + name: "with labels", + metric: "test_metric", + labels: map[string]string{"env": "prod", "host": "server1"}, + value: 42.5, + timestamp: time.Now(), + wantNil: false, + }, + { + name: "nil labels initialized", + metric: "test_metric", + labels: nil, + value: 100, + timestamp: time.Now(), + wantNil: false, + }, + { + name: "empty labels", + metric: "test_metric", + labels: map[string]string{}, + value: 0, + timestamp: time.Now(), + wantNil: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + sample := NewSample(tt.metric, tt.labels, tt.value, tt.timestamp) + + if sample.MetricName != tt.metric { + t.Errorf("MetricName = %v, want %v", sample.MetricName, tt.metric) + } + if sample.Value != tt.value { + t.Errorf("Value = %v, want %v", sample.Value, tt.value) + } + if sample.Labels == nil { + t.Error("Labels should never be nil") + } + if !sample.Timestamp.Equal(tt.timestamp) { + t.Errorf("Timestamp = %v, want %v", sample.Timestamp, tt.timestamp) + } + }) + } +} + +func TestSample_Age(t *testing.T) { + tests := []struct { + name string + sample Sample + wantNear time.Duration + }{ + { + name: "recent sample", + sample: Sample{ + MetricName: "test", + Timestamp: time.Now().Add(-5 * time.Minute), + }, + wantNear: 5 * time.Minute, + }, + { + name: "old sample", + sample: Sample{ + MetricName: "test", + Timestamp: time.Now().Add(-1 * time.Hour), + }, + wantNear: 1 * time.Hour, + }, + { + name: "very recent", + sample: Sample{ + MetricName: "test", + Timestamp: time.Now().Add(-10 * time.Second), + }, + wantNear: 10 * time.Second, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + age := tt.sample.Age() + // Allow 1 second tolerance for test execution time + if age < tt.wantNear-time.Second || age > tt.wantNear+time.Second { + t.Errorf("Age() = %v, want near %v", age, tt.wantNear) + } + }) + } +} + +func TestSample_IsRecent(t *testing.T) { + threshold := 5 * time.Minute + + tests := []struct { + name string + sample Sample + threshold time.Duration + want bool + }{ + { + name: "within threshold", + sample: Sample{ + MetricName: "test", + Timestamp: time.Now().Add(-2 * time.Minute), + }, + threshold: threshold, + want: true, + }, + { + name: "beyond threshold", + sample: Sample{ + MetricName: "test", + Timestamp: time.Now().Add(-10 * time.Minute), + }, + threshold: threshold, + want: false, + }, + { + name: "exactly at threshold", + sample: Sample{ + MetricName: "test", + Timestamp: time.Now().Add(-5 * time.Minute), + }, + threshold: threshold, + want: false, + }, + { + name: "very recent", + sample: Sample{ + MetricName: "test", + Timestamp: time.Now().Add(-10 * time.Second), + }, + threshold: threshold, + want: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.sample.IsRecent(tt.threshold); got != tt.want { + t.Errorf("IsRecent() = %v, want %v (age: %v)", got, tt.want, tt.sample.Age()) + } + }) + } +} diff --git a/internal/parser/csv.go b/internal/parser/csv.go new file mode 100644 index 0000000..64d16e5 --- /dev/null +++ b/internal/parser/csv.go @@ -0,0 +1,101 @@ +package parser + +import ( + "context" + "encoding/csv" + "fmt" + "io" + "strconv" + "strings" + "time" + + "epimetheus/internal/metrics" +) + +// CSVParser parses metrics from CSV format +type CSVParser struct{} + +// NewCSVParser creates a new CSV parser +func NewCSVParser() *CSVParser { + return &CSVParser{} +} + +// Parse reads metrics from CSV format +// Format: metric_name,label1=value1;label2=value2,value,timestamp_ms +func (p *CSVParser) Parse(ctx context.Context, reader io.Reader) ([]metrics.Sample, error) { + var samples []metrics.Sample + + csvReader := csv.NewReader(reader) + csvReader.Comment = '#' + + lineNum := 0 + for { + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + + record, err := csvReader.Read() + if err == io.EOF { + break + } + if err != nil { + return nil, fmt.Errorf("line %d: %w", lineNum, err) + } + lineNum++ + + if len(record) < 3 { + continue // Skip invalid records + } + + sample, err := p.parseRecord(record, lineNum) + if err != nil { + continue // Skip records with errors + } + + samples = append(samples, sample) + } + + return samples, nil +} + +func (p *CSVParser) parseRecord(record []string, lineNum int) (metrics.Sample, error) { + metricName := strings.TrimSpace(record[0]) + if metricName == "" { + return metrics.Sample{}, fmt.Errorf("empty metric name") + } + + labels := parseLabels(record[1]) + + value, err := strconv.ParseFloat(strings.TrimSpace(record[2]), 64) + if err != nil { + return metrics.Sample{}, fmt.Errorf("invalid value: %w", err) + } + + timestamp := time.Now() + if len(record) > 3 && record[3] != "" { + timestampMs, err := strconv.ParseInt(strings.TrimSpace(record[3]), 10, 64) + if err == nil { + timestamp = time.UnixMilli(timestampMs) + } + } + + return metrics.NewSample(metricName, labels, value, timestamp), nil +} + +func parseLabels(labelStr string) map[string]string { + labels := make(map[string]string) + if labelStr == "" { + return labels + } + + labelPairs := strings.Split(labelStr, ";") + for _, pair := range labelPairs { + parts := strings.SplitN(pair, "=", 2) + if len(parts) == 2 { + labels[strings.TrimSpace(parts[0])] = strings.TrimSpace(parts[1]) + } + } + return labels +} diff --git a/internal/parser/csv_test.go b/internal/parser/csv_test.go new file mode 100644 index 0000000..ffe9034 --- /dev/null +++ b/internal/parser/csv_test.go @@ -0,0 +1,175 @@ +package parser + +import ( + "context" + "strings" + "testing" + "time" +) + +func TestCSVParser_Parse(t *testing.T) { + tests := []struct { + name string + input string + wantCount int + wantErr bool + }{ + { + name: "valid single line", + input: `test_metric,env=prod;host=server1,42.5,1234567890000`, + wantCount: 1, + wantErr: false, + }, + { + name: "multiple lines", + input: `metric1,label1=value1,100,1234567890000 +metric2,label2=value2,200,1234567891000 +metric3,label3=value3,300,1234567892000`, + wantCount: 3, + wantErr: false, + }, + { + name: "with comments", + input: `# This is a comment +metric1,env=test,50,1234567890000 +# Another comment +metric2,env=prod,75,1234567891000`, + wantCount: 2, + wantErr: false, + }, + { + name: "no timestamp defaults to now", + input: `metric1,env=test,100`, + wantCount: 1, + wantErr: false, + }, + { + name: "no labels", + input: `metric1,,100,1234567890000`, + wantCount: 1, + wantErr: false, + }, + { + name: "empty input", + input: "", + wantCount: 0, + wantErr: false, + }, + { + name: "invalid line causes error", + input: `metric1,env=test,100,1234567890000 +invalid +metric2,env=prod,200,1234567891000`, + wantCount: 0, + wantErr: true, + }, + { + name: "invalid value skipped", + input: `metric1,env=test,not_a_number,1234567890000 +metric2,env=prod,200,1234567891000`, + wantCount: 1, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + parser := NewCSVParser() + reader := strings.NewReader(tt.input) + ctx := context.Background() + + samples, err := parser.Parse(ctx, reader) + + if (err != nil) != tt.wantErr { + t.Errorf("Parse() error = %v, wantErr %v", err, tt.wantErr) + return + } + if len(samples) != tt.wantCount { + t.Errorf("Parse() returned %d samples, want %d", len(samples), tt.wantCount) + } + }) + } +} + +func TestCSVParser_ParseLabels(t *testing.T) { + tests := []struct { + name string + input string + want map[string]string + }{ + { + name: "single label", + input: "env=prod", + want: map[string]string{"env": "prod"}, + }, + { + name: "multiple labels", + input: "env=prod;host=server1;region=us-west", + want: map[string]string{"env": "prod", "host": "server1", "region": "us-west"}, + }, + { + name: "empty string", + input: "", + want: map[string]string{}, + }, + { + name: "invalid label format skipped", + input: "env=prod;invalid;host=server1", + want: map[string]string{"env": "prod", "host": "server1"}, + }, + { + name: "with spaces", + input: " env = prod ; host = server1 ", + want: map[string]string{"env": "prod", "host": "server1"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := parseLabels(tt.input) + if len(got) != len(tt.want) { + t.Errorf("parseLabels() returned %d labels, want %d", len(got), len(tt.want)) + } + for k, v := range tt.want { + if got[k] != v { + t.Errorf("parseLabels()[%s] = %v, want %v", k, got[k], v) + } + } + }) + } +} + +func TestCSVParser_ParseWithContext(t *testing.T) { + t.Run("context cancellation", func(t *testing.T) { + parser := NewCSVParser() + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel immediately + + input := strings.NewReader(`metric1,env=test,100,1234567890000`) + _, err := parser.Parse(ctx, input) + + if err != context.Canceled { + t.Errorf("Expected context.Canceled error, got %v", err) + } + }) +} + +func TestCSVParser_ParseTimestamp(t *testing.T) { + parser := NewCSVParser() + input := `metric1,env=test,100,1234567890000` + reader := strings.NewReader(input) + ctx := context.Background() + + samples, err := parser.Parse(ctx, reader) + if err != nil { + t.Fatalf("Parse() error = %v", err) + } + if len(samples) != 1 { + t.Fatalf("Expected 1 sample, got %d", len(samples)) + } + + expectedTime := time.UnixMilli(1234567890000) + if !samples[0].Timestamp.Equal(expectedTime) { + t.Errorf("Timestamp = %v, want %v", samples[0].Timestamp, expectedTime) + } +} diff --git a/internal/parser/json.go b/internal/parser/json.go new file mode 100644 index 0000000..3b8c2e8 --- /dev/null +++ b/internal/parser/json.go @@ -0,0 +1,62 @@ +package parser + +import ( + "context" + "encoding/json" + "fmt" + "io" + "time" + + "epimetheus/internal/metrics" +) + +// JSONParser parses metrics from JSON format +type JSONParser struct{} + +// NewJSONParser creates a new JSON parser +func NewJSONParser() *JSONParser { + return &JSONParser{} +} + +type jsonSample struct { + Metric string `json:"metric"` + Labels map[string]string `json:"labels"` + Value float64 `json:"value"` + TimestampMs int64 `json:"timestamp_ms,omitempty"` +} + +// Parse reads metrics from JSON format +func (p *JSONParser) Parse(ctx context.Context, reader io.Reader) ([]metrics.Sample, error) { + var rawSamples []jsonSample + + decoder := json.NewDecoder(reader) + if err := decoder.Decode(&rawSamples); err != nil { + return nil, fmt.Errorf("failed to parse JSON: %w", err) + } + + samples := make([]metrics.Sample, 0, len(rawSamples)) + for _, raw := range rawSamples { + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + + if raw.Metric == "" { + continue + } + + timestamp := time.Now() + if raw.TimestampMs > 0 { + timestamp = time.UnixMilli(raw.TimestampMs) + } + + if raw.Labels == nil { + raw.Labels = make(map[string]string) + } + + samples = append(samples, metrics.NewSample(raw.Metric, raw.Labels, raw.Value, timestamp)) + } + + return samples, nil +} diff --git a/internal/parser/json_test.go b/internal/parser/json_test.go new file mode 100644 index 0000000..d521942 --- /dev/null +++ b/internal/parser/json_test.go @@ -0,0 +1,177 @@ +package parser + +import ( + "context" + "strings" + "testing" + "time" +) + +func TestJSONParser_Parse(t *testing.T) { + tests := []struct { + name string + input string + wantCount int + wantErr bool + }{ + { + name: "valid single sample", + input: `[ + { + "metric": "test_metric", + "labels": {"env": "prod", "host": "server1"}, + "value": 42.5, + "timestamp_ms": 1234567890000 + } + ]`, + wantCount: 1, + wantErr: false, + }, + { + name: "multiple samples", + input: `[ + {"metric": "metric1", "labels": {"env": "prod"}, "value": 100, "timestamp_ms": 1234567890000}, + {"metric": "metric2", "labels": {"env": "test"}, "value": 200, "timestamp_ms": 1234567891000}, + {"metric": "metric3", "labels": {"env": "dev"}, "value": 300, "timestamp_ms": 1234567892000} + ]`, + wantCount: 3, + wantErr: false, + }, + { + name: "no timestamp defaults to now", + input: `[ + {"metric": "test_metric", "labels": {"env": "prod"}, "value": 100} + ]`, + wantCount: 1, + wantErr: false, + }, + { + name: "no labels", + input: `[ + {"metric": "test_metric", "value": 100, "timestamp_ms": 1234567890000} + ]`, + wantCount: 1, + wantErr: false, + }, + { + name: "empty metric skipped", + input: `[ + {"metric": "", "labels": {"env": "prod"}, "value": 100}, + {"metric": "valid_metric", "labels": {"env": "test"}, "value": 200} + ]`, + wantCount: 1, + wantErr: false, + }, + { + name: "empty array", + input: `[]`, + wantCount: 0, + wantErr: false, + }, + { + name: "invalid json", + input: `{not valid json}`, + wantCount: 0, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + parser := NewJSONParser() + reader := strings.NewReader(tt.input) + ctx := context.Background() + + samples, err := parser.Parse(ctx, reader) + + if (err != nil) != tt.wantErr { + t.Errorf("Parse() error = %v, wantErr %v", err, tt.wantErr) + return + } + if len(samples) != tt.wantCount { + t.Errorf("Parse() returned %d samples, want %d", len(samples), tt.wantCount) + } + }) + } +} + +func TestJSONParser_ParseWithContext(t *testing.T) { + t.Run("context check during parse", func(t *testing.T) { + parser := NewJSONParser() + ctx, cancel := context.WithCancel(context.Background()) + + // Create valid input with empty metrics that will be filtered + input := `[ + {"metric": "", "value": 1}, + {"metric": "", "value": 2}, + {"metric": "", "value": 3} + ]` + + cancel() // Cancel before parsing + + reader := strings.NewReader(input) + _, err := parser.Parse(ctx, reader) + + // Context cancellation should be detected during sample processing + if err != context.Canceled { + // Note: JSON decoder may finish before context is checked + // This test verifies context support exists, but timing is not guaranteed + t.Logf("Got error: %v (context may not be checked until after JSON decode)", err) + } + }) +} + +func TestJSONParser_ParseTimestamp(t *testing.T) { + parser := NewJSONParser() + input := `[{"metric": "test_metric", "value": 100, "timestamp_ms": 1234567890000}]` + reader := strings.NewReader(input) + ctx := context.Background() + + samples, err := parser.Parse(ctx, reader) + if err != nil { + t.Fatalf("Parse() error = %v", err) + } + if len(samples) != 1 { + t.Fatalf("Expected 1 sample, got %d", len(samples)) + } + + expectedTime := time.UnixMilli(1234567890000) + if !samples[0].Timestamp.Equal(expectedTime) { + t.Errorf("Timestamp = %v, want %v", samples[0].Timestamp, expectedTime) + } +} + +func TestJSONParser_ParseLabels(t *testing.T) { + parser := NewJSONParser() + input := `[{ + "metric": "test_metric", + "labels": {"env": "prod", "host": "server