add main

1 files changed, 390 insertions, 0 deletions

diff --git a/cmd/epimetheus/main.go b/cmd/epimetheus/main.go
new file mode 100644
index 0000000..4f89918
--- /dev/null
+++ b/cmd/epimetheus/main.go
@@ -0,0 +1,390 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"math/rand"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
+	"epimetheus/internal/config"
+	"epimetheus/internal/ingester"
+	"epimetheus/internal/metrics"
+	"epimetheus/internal/parser"
+	"epimetheus/internal/version"
+	"epimetheus/internal/watcher"
+)
+
+func main() {
+	cfg := parseFlags()
+
+	rand.Seed(time.Now().UnixNano())
+
+	ctx, cancel := createContextWithSignalHandler()
+	defer cancel()
+
+	if err := run(ctx, cfg); err != nil {
+		log.Fatalf("Error: %v", err)
+	}
+}
+
+// parseFlags parses command-line flags and returns a Config.
+func parseFlags() config.Config {
+	cfg := config.NewConfig()
+
+	showVersion := flag.Bool("version", false, "Print version and exit")
+	mode := flag.String("mode", "realtime", "Mode: realtime, historic, backfill, auto, or watch")
+	pushgatewayURL := flag.String("pushgateway", cfg.PushgatewayURL, "Pushgateway URL for realtime mode")
+	prometheusURL := flag.String("prometheus", cfg.PrometheusURL, "Prometheus remote write URL for historic mode")
+	jobName := flag.String("job", cfg.JobName, "Job name for metrics")
+	continuous := flag.Bool("continuous", false, "For realtime mode: push continuously every 15s")
+
+	hoursAgo := flag.Int("hours-ago", cfg.HoursAgo, "For historic mode: how many hours ago (single datapoint)")
+	startHours := flag.Int("start-hours", cfg.StartHours, "For backfill: start time in hours ago")
+	endHours := flag.Int("end-hours", cfg.EndHours, "For backfill: end time in hours ago")
+	interval := flag.Int("interval", cfg.Interval, "For backfill: interval between datapoints in hours")
+
+	inputFile := flag.String("file", "", "For auto/watch mode: input file(s) with metrics (comma-separated for multiple)")
+	inputFormat := flag.String("format", cfg.InputFormat, "For auto mode: input format (csv or json)")
+	metricName := flag.String("metric-name", "", "For watch mode: metric name (optional, auto-derived from filename)")
+	resolveIPLabels := flag.String("resolve-ip-labels", "", "For watch mode: comma-separated list of additional IP labels to resolve via DNS (default: ip)")
+	clickhouseURL := flag.String("clickhouse", "", "For watch mode: ClickHouse HTTP URL (e.g. http://localhost:8123) to also ingest metrics")
+	clickhouseTable := flag.String("clickhouse-table", "epimetheus_metrics", "For watch mode: ClickHouse table name")
+
+	flag.Parse()
+
+	if *showVersion {
+		fmt.Printf("epimetheus version %s\n", version.Version)
+		os.Exit(0)
+	}
+
+	cfg.Mode = config.Mode(*mode)
+	cfg.PushgatewayURL = *pushgatewayURL
+	cfg.PrometheusURL = *prometheusURL
+	cfg.JobName = *jobName
+	cfg.Continuous = *continuous
+	cfg.HoursAgo = *hoursAgo
+	cfg.StartHours = *startHours
+	cfg.EndHours = *endHours
+	cfg.Interval = *interval
+	cfg.InputFile = *inputFile
+	cfg.InputFormat = *inputFormat
+	cfg.MetricName = *metricName
+	cfg.ClickHouseURL = *clickhouseURL
+	cfg.ClickHouseTable = *clickhouseTable
+
+	// Parse resolve-ip-labels flag
+	cfg.ResolveIPLabels = []string{"ip"} // Always include default
+	if *resolveIPLabels != "" {
+		additionalLabels := strings.Split(*resolveIPLabels, ",")
+		for _, label := range additionalLabels {
+			label = strings.TrimSpace(label)
+			if label != "" && label != "ip" { // Avoid duplicates
+				cfg.ResolveIPLabels = append(cfg.ResolveIPLabels, label)
+			}
+		}
+	}
+
+	return cfg
+}
+
+// createContextWithSignalHandler creates a context that cancels on interrupt signals.
+func createContextWithSignalHandler() (context.Context, context.CancelFunc) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+
+	go func() {
+		<-sigChan
+		log.Printf("\nReceived interrupt signal, shutting down...")
+		cancel()
+	}()
+
+	return ctx, cancel
+}
+
+// run executes the appropriate mode based on configuration.
+func run(ctx context.Context, cfg config.Config) error {
+	switch cfg.Mode {
+	case config.ModeRealtime:
+		return runRealtimeMode(ctx, cfg)
+	case config.ModeHistoric:
+		return runHistoricMode(ctx, cfg)
+	case config.ModeBackfill:
+		return runBackfillMode(ctx, cfg)
+	case config.ModeAuto:
+		return runAutoMode(ctx, cfg)
+	case config.ModeWatch:
+		return runWatchMode(ctx, cfg)
+	default:
+		return fmt.Errorf("unknown mode: %s (use realtime, historic, backfill, auto, or watch)", cfg.Mode)
+	}
+}
+
+// runRealtimeMode runs the realtime ingestion mode.
+func runRealtimeMode(ctx context.Context, cfg config.Config) error {
+	log.Printf("Starting Prometheus metrics pusher in REALTIME mode")
+	log.Printf("Pushgateway URL: %s", cfg.PushgatewayURL)
+	log.Printf("Job name: %s", cfg.JobName)
+
+	collectors := metrics.NewCollectors()
+	pushgateway := ingester.NewPushgatewayIngester()
+
+	if err := pushgateway.Ingest(ctx, collectors, cfg.PushgatewayURL, cfg.JobName); err != nil {
+		return fmt.Errorf("failed to push metrics: %w", err)
+	}
+	log.Printf("Successfully pushed metrics to Pushgateway")
+
+	if cfg.Continuous {
+		return runContinuousMode(ctx, pushgateway, collectors, cfg)
+	}
+
+	return nil
+}
+
+// runContinuousMode pushes metrics continuously every 15 seconds.
+func runContinuousMode(ctx context.Context, pushgateway ingester.PushgatewayIngester, collectors metrics.Collectors, cfg config.Config) error {
+	ticker := time.NewTicker(15 * time.Second)
+	defer ticker.Stop()
+
+	log.Printf("Continuous mode: pushing metrics every 15 seconds. Press Ctrl+C to stop.")
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			if err := pushgateway.Ingest(ctx, collectors, cfg.PushgatewayURL, cfg.JobName); err != nil {
+				log.Printf("Error pushing metrics: %v", err)
+			} else {
+				log.Printf("Successfully pushed metrics to Pushgateway")
+			}
+		}
+	}
+}
+
+// runHistoricMode runs the historic ingestion mode.
+func runHistoricMode(ctx context.Context, cfg config.Config) error {
+	remoteWrite := ingester.NewRemoteWriteIngester()
+	return remoteWrite.IngestHistoric(ctx, cfg.PrometheusURL, cfg.HoursAgo)
+}
+
+// runBackfillMode runs the backfill ingestion mode.
+func runBackfillMode(ctx context.Context, cfg config.Config) error {
+	remoteWrite := ingester.NewRemoteWriteIngester()
+	return remoteWrite.Backfill(ctx, cfg.PrometheusURL, cfg.StartHours, cfg.EndHours, cfg.Interval)
+}
+
+// runAutoMode runs the auto ingestion mode.
+func runAutoMode(ctx context.Context, cfg config.Config) error {
+	log.Printf("🤖 AUTO mode: Automatically detecting timestamp age and choosing ingestion method")
+
+	samples, err := loadSamples(ctx, cfg)
+	if err != nil {
+		return err
+	}
+
+	logFileSource(cfg)
+
+	collectors := metrics.NewCollectors()
+	autoIngester := ingester.NewAutoIngester(collectors)
+
+	return autoIngester.Ingest(ctx, samples, cfg)
+}
+
+// loadSamples loads samples from file or stdin based on configuration.
+func loadSamples(ctx context.Context, cfg config.Config) ([]metrics.Sample, error) {
+	if cfg.InputFile != "" {
+		return parser.ParseFile(ctx, cfg.InputFile, cfg.InputFormat)
+	}
+	return parser.ParseStdin(ctx, cfg.InputFormat)
+}
+
+// logFileSource logs the source of the input data.
+func logFileSource(cfg config.Config) {
+	if cfg.InputFile != "" {
+		log.Printf("📁 Reading metrics from: %s (format: %s)", cfg.InputFile, cfg.InputFormat)
+	} else {
+		log.Printf("📥 Reading metrics from stdin (format: %s)", cfg.InputFormat)
+	}
+}
+
+// runWatchMode runs the watch mode for monitoring CSV file changes.
+// Supports watching multiple files by providing comma-separated list.
+func runWatchMode(ctx context.Context, cfg config.Config) error {
+	if cfg.InputFile == "" {
+		return fmt.Errorf("watch mode requires -file flag")
+	}
+
+	// Parse comma-separated file list
+	files := strings.Split(cfg.InputFile, ",")
+	for i := range files {
+		files[i] = strings.TrimSpace(files[i])
+	}
+
+	log.Printf("👁️  WATCH mode: Monitoring %d file(s)", len(files))
+	if cfg.PrometheusURL != "" {
+		log.Printf("Prometheus URL: %s", cfg.PrometheusURL)
+	}
+	if cfg.ClickHouseURL != "" {
+		log.Printf("ClickHouse URL: %s (table: %s)", cfg.ClickHouseURL, cfg.ClickHouseTable)
+	}
+	log.Printf("Job name: %s", cfg.JobName)
+	log.Printf("Polling interval: 1 second")
+
+	// Create ingesters (shared across all files)
+	var remoteWrite *ingester.RemoteWriteIngester
+	if cfg.PrometheusURL != "" {
+		rw := ingester.NewRemoteWriteIngester()
+		remoteWrite = &rw
+	}
+
+	var clickhouse *ingester.ClickHouseIngester
+	if cfg.ClickHouseURL != "" {
+		ch := ingester.NewClickHouseIngester(cfg.ClickHouseURL, cfg.ClickHouseTable)
+		clickhouse = &ch
+	}
+
+	if remoteWrite == nil && clickhouse == nil {
+		return fmt.Errorf("watch mode requires at least one of -prometheus or -clickhouse")
+	}
+
+	// Use WaitGroup to manage multiple watchers
+	var wg sync.WaitGroup
+	errChan := make(chan error, len(files))
+
+	// Start a watcher for each file
+	for _, filePath := range files {
+		// Determine metric name: use -metric-name if provided (and only one file),
+		// otherwise extract from filename
+		metricName := cfg.MetricName
+		if metricName == "" || len(files) > 1 {
+			metricName = extractMetricNameFromFile(filePath)
+		}
+
+		log.Printf("📁 Watching: %s → metric: %s", filePath, metricName)
+
+		wg.Add(1)
+		go func(file, metric string) {
+			defer wg.Done()
+			if err := watchSingleFile(ctx, file, metric, remoteWrite, clickhouse, cfg); err != nil {
+				errChan <- fmt.Errorf("watcher for %s failed: %w", file, err)
+			}
+		}(filePath, metricName)
+	}
+
+	log.Printf("✅ Watching for file changes. Press Ctrl+C to stop.")
+
+	// Wait for all watchers to finish or context cancellation
+	go func() {
+		wg.Wait()
+		close(errChan)
+	}()
+
+	// Collect any errors
+	var errors []error
+	for err := range errChan {
+		errors = append(errors, err)
+		log.Printf("⚠️  Watcher error: %v", err)
+	}
+
+	if len(errors) > 0 {
+		return fmt.Errorf("watch mode completed with %d error(s)", len(errors))
+	}
+
+	return nil
+}
+
+// watchSingleFile watches a single file for changes
+func watchSingleFile(ctx context.Context, filePath string, metricName string, remoteWrite *ingester.RemoteWriteIngester, clickhouse *ingester.ClickHouseIngester, cfg config.Config) error {
+	// Create file watcher
+	fw := watcher.NewFileWatcher(filePath, 1*time.Second)
+	changeChan, err := fw.Watch(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to start file watcher: %w", err)
+	}
+
+	// Process file on each change
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case _, ok := <-changeChan:
+			if !ok {
+				// Channel closed, watcher stopped
+				return nil
+			}
+
+			if err := processWatchedFile(ctx, filePath, metricName, fw.GetLastModTime(), remoteWrite, clickhouse, cfg); err != nil {
+				log.Printf("⚠️  Error processing %s: %v", filePath, err)
+				// Continue watching despite errors
+				continue
+			}
+		}
+	}
+}
+
+// extractMetricNameFromFile extracts metric name from filename
+// e.g., "/path/to/blockstore.csv" -> "blockstore"
+func extractMetricNameFromFile(filePath string) string {
+	base := filepath.Base(filePath)
+	name := strings.TrimSuffix(base, filepath.Ext(base))
+	return name
+}
+
+// processWatchedFile reads and ingests metrics from the watched file
+func processWatchedFile(ctx context.Context, filePath string, metricName string, modTime time.Time, remoteWrite *ingester.RemoteWriteIngester, clickhouse *ingester.ClickHouseIngester, cfg config.Config) error {
+	filename := filepath.Base(filePath)
+	log.Printf("📊 [%s] File changed, processing metrics (metric: %s)...", filename, metricName)
+
+	// Open the file
+	file, err := os.Open(filePath)
+	if err != nil {
+		return fmt.Errorf("failed to open file: %w", err)
+	}
+	defer file.Close()
+
+	// Parse using tabular CSV parser
+	tabularParser := parser.NewTabularCSVParser(metricName, modTime, cfg.ResolveIPLabels)
+	samples, err := tabularParser.Parse(ctx, file)
+	if err != nil {
+		return fmt.Errorf("failed to parse CSV: %w", err)
+	}
+
+	if len(samples) == 0 {
+		log.Printf("⚠️  [%s] No samples found in file", filename)
+		return nil
+	}
+
+	log.Printf("📦 [%s] Parsed %d samples (timestamp: %s)", filename, len(samples), modTime.Format(time.RFC3339))
+
+	// Push to Prometheus via Remote Write API (preserves timestamp)
+	if remoteWrite != nil {
+		if err := remoteWrite.Ingest(ctx, samples, cfg.PrometheusURL); err != nil {
+			return fmt.Errorf("failed to push metrics to Prometheus: %w", err)
+		}
+		log.Printf("✅ [%s] Successfully pushed %d samples to Prometheus", filename, len(samples))
+	}
+
+	// Push to ClickHouse
+	if clickhouse != nil {
+		if err := clickhouse.Ingest(ctx, samples); err != nil {
+			return fmt.Errorf("failed to push metrics to ClickHouse: %w", err)
+		}
+		log.Printf("✅ [%s] Successfully pushed %d samples to ClickHouse", filename, len(samples))
+	}
+	
+	// Save DNS cache to disk after successful ingestion
+	tabularParser.SaveDNSCache()
+	
+	return nil
+}
+