Convert CSV to OpenMetrics

CSV_to_OpenMetrics.md

I recently was looking into ways to backfill historical data into Prometheus from a historical source.

I was hoping to use Prometheus's backfilling feature but it requires data to be in OpenMetrics format, which I was struggling to find good tooling for.

I wrote this snippet in go to convert data in CSV format to OpenMetrics format.

It combines the CSV parsing from VictoriaMetrics CSV importing, and the OpenMetrics capabilities from Prometheus' expfmt package. VictoriaMetrics code is used to parse the CSV into an in-memory representation, func ConvertVMToProm() converts the in-memory representation to something expfmt understands, and expfmt writes the file.

Note -help flag is broken due to extra flags.

This isn't tested, and isn't in a production-ready state. Also it only supports untyped point metrics.

main.go

// Copyright 2025 Google LLC.
// SPDX-License-Identifier: Apache-2.0
package main

import (
	"flag"
	"fmt"
	"log"
	"os"

	"github.com/VictoriaMetrics/metrics"
	model "github.com/prometheus/client_model/go"
	"github.com/prometheus/common/expfmt"

	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport"
)
var format, inFile, outFile string
func realMain() error {
	flag.StringVar(&format, "format", "", "format specifier")
	flag.StringVar(&inFile, "in", "", "input file")
	flag.StringVar(&outFile, "out", "", "output file -- optional")
	flag.Parse()
	if format == "" || inFile == "" {
		fmt.Println("format", format)
		fmt.Println("infile", inFile)
		return fmt.Errorf("missing required input")
	}
	if outFile == "" {
		outFile = inFile + ".om"
	}
	contents, err := os.ReadFile(inFile)
	if err != nil {
		return fmt.Errorf("bad file: %w", err)
	}
	var rows csvimport.Rows
	col, err := csvimport.ParseColumnDescriptors(format)
	if err != nil {
		return fmt.Errorf("bad column descriptor: %w", err)
	}
	rows.Unmarshal(string(contents), col)
	if failedRows := metrics.GetOrCreateCounter(`vm_rows_invalid_total{type="csvimport"}`).Get(); failedRows > 0 {
		return fmt.Errorf("could not parse %d csv rows", failedRows)
	}
	mf, err := ConvertVMToProm(&rows)
	if err != nil {
		return fmt.Errorf("error converting victoriametrics to prometeheus: %w", err)
	}
	f, err := os.Create(outFile)
	if err != nil {
		return fmt.Errorf("error creating output file: %w", err)
	}
	defer f.Close()
	_, err = expfmt.MetricFamilyToOpenMetrics(f, mf)
	if err != nil {
		return fmt.Errorf("failed to write OM to file: %w", err)
	}
	fmt.Fprintln(f, "# EOF") // # EOF\n is required at the end of OpenMetrics files and MetricFamilyToOpenMetrics does not include.
	if err != nil {
		return fmt.Errorf("failed to write EOF file: %w", err)
	}
	return nil
}
func main() {
	if err := realMain(); err != nil {
		log.Fatalln(err)
	}
}
func ConvertVMToProm(rows *csvimport.Rows) (*model.MetricFamily, error) {
	if rows == nil {
		return nil, fmt.Errorf("rows is nil")
	}
	if len(rows.Rows) == 0 {
		return nil, fmt.Errorf("rows is empty")
	}
	points := []*model.Metric{}
	for i, row := range rows.Rows {
		if row.Metric != rows.Rows[0].Metric {
			return nil, fmt.Errorf("row %d had metric name %d, but row 0 had metric name %d", i, row.Metric, rows.Rows[0].Metric)
		}
		labels := []*model.LabelPair{}
		for _, tag := range row.Tags {
			labels = append(labels, &model.LabelPair{
				Name:  &tag.Key,
				Value: &tag.Value,
			})
		}
		metric := model.Metric{
			Label:       labels,
			Untyped:     &model.Untyped{Value: &row.Value},
			TimestampMs: &row.Timestamp, // verified default is in ms https://github.com/VictoriaMetrics/VictoriaMetrics/blob/6828cca5a6fe1e8e4e8ada78f4376fa72ea1f624/lib/protoparser/csvimport/column_descriptor.go#L153
		}
		points = append(points, &metric)
	}
	mf := model.MetricFamily{
		Name:   &rows.Rows[0].Metric,
		Type:   model.MetricType_UNTYPED.Enum(),
		Metric: points,
	}
	return &mf, nil
}