main.go

package main

import (
	"context"
	"flag"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/oklog/oklog/pkg/group"
	"github.com/prometheus/tsdb"
)

func main() {
	v1Path := flag.String("v1-path", "./data-v1", "Path to the v1 storage directory.")
	v2Path := flag.String("v2-path", "./data-v2", "Path to the v2 storage directory.")
	lookback := flag.Duration("lookback", 15*24*time.Hour, "How far back to start when exporting old data.")
	step := flag.Duration("step", 15*time.Minute, "How much data to load at once.")
	parallelism := flag.Int("parallelism", 1, "How many series to migrate at the same time.")
	skipUnknownLabels := flag.Bool("skip-unknown-labels", false, "Skip unknown metrics/labels instead of failing migration.")
	flag.Parse()

	logger := log.NewSyncLogger(log.NewLogfmtLogger(os.Stderr))

	if *parallelism < 1 {
		level.Error(logger).Log("err", "-parallelism must be set to at least 1")
		os.Exit(1)
	}

	v2Storage, err := tsdb.Open(*v2Path, logger, nil, &tsdb.Options{
		RetentionDuration: 999999 * 24 * 60 * 60 * 1000,
		BlockRanges:       []int64{2 * 60 * 60 * 1000},
	})
	if err != nil {
		level.Error(logger).Log("msg", "error starting v2 storage", "err", err)
		os.Exit(1)
	}
	defer v2Storage.Close()

	migrator := storageMigrator{
		v1Path:            *v1Path,
		v2Storage:         v2Storage,
		parallelism:       *parallelism,
		skipUnknownLabels: *skipUnknownLabels,
	}

	var g group.Group
	{
		// Interrupt handler.
		term := make(chan os.Signal)
		signal.Notify(term, os.Interrupt, syscall.SIGTERM)
		cancel := make(chan struct{})
		g.Add(
			func() error {
				select {
				case <-term:
					level.Warn(logger).Log("msg", "received SIGTERM, exiting gracefully...")
				case <-cancel:
					break
				}
				return nil
			},
			func(err error) {
				close(cancel)
			},
		)
	}
	{
		// Migrator.
		ctx, cancel := context.WithCancel(context.Background())
		g.Add(
			func() error {
				if err := migrator.migrate(ctx, *lookback, *step); err != nil {
					return fmt.Errorf("error migrating storage: %s", err)
				}
				return nil
			},
			func(err error) {
				cancel()
			},
		)
	}
	if err := g.Run(); err != nil {
		level.Error(logger).Log("err", err)
	}
	level.Info(logger).Log("msg", "Done.")
}