vmctl: add support of the migration from the influx v2

app/vmctl/flags.go

@@ -311,6 +311,8 @@ var (
 
 const (
 	influxAddr                      = "influx-addr"
+	influxVersion                   = "influx-version"
+	influxToken                     = "influx-token"
 	influxUser                      = "influx-user"
 	influxPassword                  = "influx-password"
 	influxDB                        = "influx-database"
@@ -337,6 +339,19 @@ var (
 			Value: "http://localhost:8086",
 			Usage: "InfluxDB server addr",
 		},
+		&cli.IntFlag{
+			Name: influxVersion,
+			Usage: "Major version of the source InfluxDB: 1 or 2.\n" +
+				"InfluxDB 2.x is migrated via its InfluxDB 1.x compatibility API, which requires -influx-token " +
+				"and a database/retention policy mapping (DBRP) pointing at the bucket to migrate.\n" +
+				"See https://docs.influxdata.com/influxdb/v2/api-guide/influxdb-1x/",
+			Value: 1,
+		},
+		&cli.StringFlag{
+			Name:    influxToken,
+			Usage:   "InfluxDB v2 API token. Requires -influx-version=2",
+			EnvVars: []string{"INFLUX_TOKEN"},
+		},
 		&cli.StringFlag{
 			Name:    influxUser,
 			Usage:   "InfluxDB user",
@@ -353,9 +368,11 @@ var (
 			Required: true,
 		},
 		&cli.StringFlag{
-			Name:  influxRetention,
-			Usage: "InfluxDB retention policy",
-			Value: "autogen",
+			Name: influxRetention,
+			Usage: "InfluxDB retention policy.\n" +
+				"Defaults to 'autogen' for -influx-version=1.\n" +
+				"For -influx-version=2 it is the retention policy of a DBRP mapping; if empty, " +
+				"the default mapping of -influx-database is used",
 		},
 		&cli.IntFlag{
 			Name:  influxChunkSize,

app/vmctl/influx/influx.go

@@ -11,6 +11,27 @@ import (
 	influx "github.com/influxdata/influxdb/client/v2"
 )
 
+const (
+	// VersionV1 is InfluxDB 1.x, queried via its native /query endpoint.
+	VersionV1 = 1
+
+	// VersionV2 is InfluxDB 2.x, queried via its InfluxDB 1.x compatibility
+	// API. It authenticates with an API token instead of a password.
+	//
+	// See https://docs.influxdata.com/influxdb/v2/api-guide/influxdb-1x/
+	VersionV2 = 2
+
+	// defaultV1CompatUser is sent as the username when migrating from InfluxDB 2.x
+	// with an API token.
+	//
+	// The InfluxDB 1.x compatibility API requires a username whenever an API token
+	// is used as the password, but the value itself is ignored.
+	// See https://docs.influxdata.com/influxdb/v2/api-guide/influxdb-1x/
+	defaultV1CompatUser = "vmctl"
+
+	defaultRetentionV1 = "autogen"
+)
+
 // Client represents a wrapper over
 // influx HTTP client
 type Client struct {
@@ -27,7 +48,9 @@ type Client struct {
 // Config contains fields required
 // for Client configuration
 type Config struct {
+	Version   int
 	Addr      string
+	Token     string
 	Username  string
 	Password  string
 	Database  string
@@ -88,10 +111,18 @@ type LabelPair struct {
 // NewClient creates and returns influx client
 // configured with passed Config
 func NewClient(cfg Config) (*Client, error) {
+	if err := cfg.validate(); err != nil {
+		return nil, err
+	}
+
+	// InfluxDB 2.x authenticates with an API token passed as the password
+	// of its InfluxDB 1.x compatibility API.
+	username, password := resolveAuth(cfg.Username, cfg.Password, cfg.Token)
+
 	c := influx.HTTPConfig{
 		Addr:      cfg.Addr,
-		Username:  cfg.Username,
-		Password:  cfg.Password,
+		Username:  username,
+		Password:  password,
 		TLSConfig: cfg.TLSConfig,
 	}
 	hc, err := influx.NewHTTPClient(c)
@@ -99,6 +130,7 @@ func NewClient(cfg Config) (*Client, error) {
 		return nil, fmt.Errorf("failed to establish conn: %w", err)
 	}
 	if _, _, err := hc.Ping(time.Second); err != nil {
+		_ = hc.Close()
 		return nil, fmt.Errorf("ping failed: %w", err)
 	}
 
@@ -110,7 +142,7 @@ func NewClient(cfg Config) (*Client, error) {
 	client := &Client{
 		Client:       hc,
 		database:     cfg.Database,
-		retention:    cfg.Retention,
+		retention:    resolveRetention(cfg.Version, cfg.Retention),
 		chunkSize:    chunkSize,
 		filterTime:   timeFilter(cfg.Filter.TimeStart, cfg.Filter.TimeEnd),
 		filterSeries: cfg.Filter.Series,
@@ -464,3 +496,52 @@ func (c *Client) do(q influx.Query) ([]queryValues, error) {
 	}
 	return parseResult(res.Results[0])
 }
+
+// resolveAuth returns the credentials to authenticate with.
+//
+// InfluxDB 2.x is queried via its InfluxDB 1.x compatibility API, which accepts
+// an API token in place of the password. Therefore a non-empty token replaces
+// the password, and a placeholder username is substituted when none is given.
+func resolveAuth(username, password, token string) (string, string) {
+	if token == "" {
+		return username, password
+	}
+	if username == "" {
+		username = defaultV1CompatUser
+	}
+	return username, token
+}
+
+// resolveRetention returns the retention policy to query.
+//
+// In InfluxDB 1.x `autogen` is the retention policy created together with a
+// database, so it is a meaningful default. In InfluxDB 2.x the retention policy
+// is one half of a DBRP mapping and its name is arbitrary, so no default can be
+// assumed: an empty value makes InfluxDB use the default mapping of the
+// database instead of failing on a non-existent one.
+func resolveRetention(version int, retention string) string {
+	if retention == "" && version == VersionV1 {
+		return defaultRetentionV1
+	}
+	return retention
+}
+
+// validate checks that the configuration is self-consistent.
+func (cfg *Config) validate() error {
+	if cfg.Version != VersionV1 && cfg.Version != VersionV2 {
+		return fmt.Errorf("unsupported InfluxDB version %d; supported versions are %d and %d",
+			cfg.Version, VersionV1, VersionV2)
+	}
+	if cfg.Version == VersionV2 && cfg.Token == "" {
+		return fmt.Errorf("-influx-token is required for InfluxDB v2")
+	}
+	if cfg.Version == VersionV1 && cfg.Token != "" {
+		return fmt.Errorf("-influx-token is only supported for InfluxDB v2; pass -influx-version=2 to use it")
+	}
+	// The database is the `db` parameter of the query API. For InfluxDB 2.x
+	// it is the database name of a DBRP mapping, which points at a bucket.
+	if cfg.Database == "" {
+		return fmt.Errorf("-influx-database cannot be empty")
+	}
+	return nil
+}

app/vmctl/influx/influx_test.go

@@ -1,6 +1,12 @@
 package influx
 
-import "testing"
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+)
 
 func TestFetchQuery(t *testing.T) {
 	f := func(s *Series, timeFilter, resultExpected string) {
@@ -126,3 +132,255 @@ func TestGetSeriesCommand(t *testing.T) {
 	f("from cpu where arch='x86'", "time >= '2020-01-01T20:07:00Z'", "show series from cpu where arch='x86' AND time >= '2020-01-01T20:07:00Z'")
 	f("from cpu where arch='x86' AND hostname='host_2753'", "time >= '2020-01-01T20:07:00Z'", "show series from cpu where arch='x86' AND hostname='host_2753' AND time >= '2020-01-01T20:07:00Z'")
 }
+
+func TestResolveAuth(t *testing.T) {
+	f := func(username, password, token, userExpected, passExpected string) {
+		t.Helper()
+
+		user, pass := resolveAuth(username, password, token)
+		if user != userExpected || pass != passExpected {
+			t.Fatalf("unexpected credentials for (username=%q, password=%q, token=%q)\ngot\n(%q, %q)\nwant\n(%q, %q)",
+				username, password, token, user, pass, userExpected, passExpected)
+		}
+	}
+
+	// InfluxDB 1.x: no token, credentials are passed through unchanged.
+	f("", "", "", "", "")
+	f("user", "pass", "", "user", "pass")
+
+	// InfluxDB 2.x: the API token is sent as the password. The username is
+	// required by the v1 compatibility API but may be any value, so a
+	// placeholder is used when the caller did not provide one.
+	f("", "", "token", defaultV1CompatUser, "token")
+
+	// An explicitly provided username is preserved.
+	f("myuser", "", "token", "myuser", "token")
+
+	// The token takes precedence over a password.
+	f("user", "pass", "token", "user", "token")
+}
+
+func TestConfigValidateSuccess(t *testing.T) {
+	f := func(cfg Config) {
+		t.Helper()
+
+		if err := cfg.validate(); err != nil {
+			t.Fatalf("unexpected error for config %+v: %s", cfg, err)
+		}
+	}
+
+	// InfluxDB 1.x with and without credentials.
+	f(Config{Version: 1, Database: "mydb"})
+	f(Config{Version: 1, Database: "mydb", Username: "user", Password: "pass"})
+
+	// InfluxDB 2.x requires a token.
+	f(Config{Version: 2, Database: "mydb", Token: "my-token"})
+}
+
+func TestConfigValidateFailure(t *testing.T) {
+	f := func(cfg Config, errStrExpected string) {
+		t.Helper()
+
+		err := cfg.validate()
+		if err == nil {
+			t.Fatalf("expecting non-nil error for config %+v", cfg)
+		}
+		if !strings.Contains(err.Error(), errStrExpected) {
+			t.Fatalf("unexpected error for config %+v\ngot\n%s\nwant it to contain\n%s", cfg, err, errStrExpected)
+		}
+	}
+
+	// unsupported versions
+	f(Config{Version: 0, Database: "mydb"}, "unsupported InfluxDB version")
+	f(Config{Version: 3, Database: "mydb"}, "unsupported InfluxDB version")
+
+	// InfluxDB 2.x without a token cannot authenticate
+	f(Config{Version: 2, Database: "mydb"}, "influx-token")
+
+	// a token is meaningless without opting into v2
+	f(Config{Version: 1, Database: "mydb", Token: "my-token"}, "influx-version=2")
+
+	// the database is mandatory for both versions, since it is the `db`
+	// parameter of the query API
+	f(Config{Version: 1}, "influx-database")
+	f(Config{Version: 2, Token: "my-token"}, "influx-database")
+}
+
+func TestResolveRetention(t *testing.T) {
+	f := func(version int, retention, resultExpected string) {
+		t.Helper()
+
+		result := resolveRetention(version, retention)
+		if result != resultExpected {
+			t.Fatalf("unexpected retention policy for (version=%d, retention=%q)\ngot\n%q\nwant\n%q",
+				version, retention, result, resultExpected)
+		}
+	}
+
+	// InfluxDB 1.x keeps its historical default.
+	f(VersionV1, "", "autogen")
+	f(VersionV1, "all_data", "all_data")
+
+	// For InfluxDB 2.x the retention policy is part of a DBRP mapping whose
+	// name is arbitrary, so no default may be assumed: an empty value lets
+	// InfluxDB pick the default mapping of the database.
+	f(VersionV2, "", "")
+	f(VersionV2, "myrp", "myrp")
+}
+
+// TestNewClientValidatesConfig ensures an invalid configuration is rejected
+// before any connection to InfluxDB is attempted.
+func TestNewClientValidatesConfig(t *testing.T) {
+	f := func(cfg Config, errStrExpected string) {
+		t.Helper()
+
+		cfg.Addr = "http://127.0.0.1:1"
+
+		c, err := NewClient(cfg)
+		if err == nil {
+			t.Fatalf("expecting non-nil error for config %+v", cfg)
+		}
+		if c != nil {
+			t.Fatalf("expecting nil client for config %+v", cfg)
+		}
+		if !strings.Contains(err.Error(), errStrExpected) {
+			t.Fatalf("unexpected error for config %+v\ngot\n%s\nwant it to contain\n%s", cfg, err, errStrExpected)
+		}
+	}
+
+	f(Config{Version: 0, Database: "mydb"}, "unsupported InfluxDB version")
+	f(Config{Version: 2, Database: "mydb"}, "influx-token")
+	f(Config{Version: 1, Database: "mydb", Token: "my-token"}, "influx-version=2")
+}
+
+// TestNewClientTokenAuth ensures the API token is used as the password when
+// migrating from InfluxDB 2.x.
+func TestNewClientTokenAuth(t *testing.T) {
+	f := func(cfg Config, userExpected, passExpected string) {
+		t.Helper()
+
+		user, pass := resolveAuth(cfg.Username, cfg.Password, cfg.Token)
+		if user != userExpected || pass != passExpected {
+			t.Fatalf("unexpected credentials for config %+v\ngot\n(%q, %q)\nwant\n(%q, %q)",
+				cfg, user, pass, userExpected, passExpected)
+		}
+	}
+
+	f(Config{Version: 2, Database: "mydb", Token: "my-token"}, defaultV1CompatUser, "my-token")
+	f(Config{Version: 1, Database: "mydb", Username: "user", Password: "pass"}, "user", "pass")
+}
+
+func TestQueryRequestAuthAndRetention(t *testing.T) {
+	f := func(cfg Config, userExpected, passExpected, rpExpected string) {
+		t.Helper()
+
+		var lastQuery queryRequest
+		s := newTestServer(t, &lastQuery)
+		defer s.Close()
+
+		cfg.Addr = s.URL
+		c, err := NewClient(cfg)
+		if err != nil {
+			t.Fatalf("unexpected error creating client: %s", err)
+		}
+
+		if _, err := c.fieldsByMeasurement(); err != nil {
+			t.Fatalf("unexpected error querying field keys: %s", err)
+		}
+
+		got := lastQuery.get()
+		if !got.seen {
+			t.Fatalf("the client did not issue a /query request")
+		}
+		if !got.hasAuth {
+			t.Fatalf("expecting basic auth to be set on the request")
+		}
+		if got.user != userExpected || got.password != passExpected {
+			t.Fatalf("unexpected credentials on the wire\ngot\n(%q, %q)\nwant\n(%q, %q)",
+				got.user, got.password, userExpected, passExpected)
+		}
+		if got.rp != rpExpected {
+			t.Fatalf("unexpected rp parameter\ngot\n%q\nwant\n%q", got.rp, rpExpected)
+		}
+		if got.db != cfg.Database {
+			t.Fatalf("unexpected db parameter\ngot\n%q\nwant\n%q", got.db, cfg.Database)
+		}
+	}
+
+	// InfluxDB 1.x: credentials are passed through and `autogen` is the
+	// default retention policy.
+	f(Config{Version: VersionV1, Database: "mydb", Username: "user", Password: "pass"},
+		"user", "pass", "autogen")
+
+	// An explicit retention policy is honoured.
+	f(Config{Version: VersionV1, Database: "mydb", Username: "user", Password: "pass", Retention: "all_data"},
+		"user", "pass", "all_data")
+
+	// InfluxDB 2.x: the API token is sent as the password with a placeholder
+	// username, and no retention policy is assumed so that the default DBRP
+	// mapping of the database applies.
+	f(Config{Version: VersionV2, Database: "mydb", Token: "my-token"},
+		defaultV1CompatUser, "my-token", "")
+
+	// An explicit retention policy names the DBRP mapping to query.
+	f(Config{Version: VersionV2, Database: "mydb", Token: "my-token", Retention: "myrp"},
+		defaultV1CompatUser, "my-token", "myrp")
+}
+
+// queryRequest holds what a /query request carried on the wire. Only the values
+// under test are kept, so nothing of the *http.Request outlives the handler.
+type queryRequest struct {
+	mu sync.Mutex
+
+	seen     bool
+	user     string
+	password string
+	hasAuth  bool
+	rp       string
+	db       string
+}
+
+func (q *queryRequest) set(r *http.Request) {
+	user, password, hasAuth := r.BasicAuth()
+	params := r.URL.Query()
+
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	q.seen = true
+	q.user, q.password, q.hasAuth = user, password, hasAuth
+	q.rp = params.Get("rp")
+	q.db = params.Get("db")
+}
+
+func (q *queryRequest) get() queryRequest {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	return queryRequest{
+		seen:     q.seen,
+		user:     q.user,
+		password: q.password,
+		hasAuth:  q.hasAuth,
+		rp:       q.rp,
+		db:       q.db,
+	}
+}
+
+func newTestServer(t *testing.T, lastQuery *queryRequest) *httptest.Server {
+	t.Helper()
+
+	const fieldKeysResponse = `{"results":[{"statement_id":0,"series":[{"name":"cpu","columns":["fieldKey","fieldType"],"values":[["value","float"]]}]}]}`
+
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("X-Influxdb-Version", "test")
+		switch r.URL.Path {
+		case "/ping":
+			w.WriteHeader(http.StatusNoContent)
+		case "/query":
+			lastQuery.set(r)
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(fieldKeysResponse))
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+}

app/vmctl/main.go

@@ -141,7 +141,9 @@ func main() {
 					}
 
 					iCfg := influx.Config{
+						Version:   c.Int(influxVersion),
 						Addr:      c.String(influxAddr),
+						Token:     c.String(influxToken),
 						Username:  c.String(influxUser),
 						Password:  c.String(influxPassword),
 						Database:  c.String(influxDB),

apptest/tests/influx_server_test.go

@@ -0,0 +1,342 @@
+package tests
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"slices"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// influxPoint is a single field value at a single timestamp of a single series,
+// as served by the mock InfluxDB server.
+type influxPoint struct {
+	Measurement string
+	Tags        map[string]string
+	Field       string
+	FieldType   string
+	Timestamp   int64 // unix seconds
+	Value       any   // float64, int64, bool or string
+}
+
+// influxRequest is a /query request recorded by the mock server, used to assert
+// how vmctl authenticates and which database and retention policy it asks for.
+type influxRequest struct {
+	Query    url.Values
+	User     string
+	Password string
+	HasAuth  bool
+}
+
+// influxMockServer implements the subset of the InfluxDB 1.x query API used by
+// `vmctl influx`: /ping plus /query serving `SHOW FIELD KEYS`, `SHOW TAG KEYS`,
+// `SHOW SERIES` and `SELECT`.
+//
+// InfluxDB 2.x is migrated through the very same API - its 1.x compatibility
+// endpoint - so one mock covers both -influx-version=1 and -influx-version=2.
+type influxMockServer struct {
+	server *httptest.Server
+	points []influxPoint
+
+	mu       sync.Mutex
+	requests []influxRequest
+}
+
+// newInfluxMockServer starts an httptest server serving the given points.
+func newInfluxMockServer(t *testing.T, points []influxPoint) *influxMockServer {
+	t.Helper()
+	s := &influxMockServer{points: points}
+	mux := http.NewServeMux()
+	mux.HandleFunc("/ping", s.handlePing)
+	mux.HandleFunc("/query", s.handleQuery)
+	s.server = httptest.NewServer(mux)
+	return s
+}
+
+func (s *influxMockServer) close() { s.server.Close() }
+
+func (s *influxMockServer) httpAddr() string { return s.server.URL }
+
+// recordQuery stores a /query request for later assertions.
+//
+// Every request is kept, not just the last one: vmctl issues several kinds of
+// query - `SHOW FIELD KEYS` unchunked, `SHOW TAG KEYS` and `SHOW SERIES`
+// chunked, and one `SELECT` per series - and the assertions verify that all of
+// them authenticate and address the database the same way. The assertions
+// cannot live in this handler because it runs on a server goroutine, where
+// t.Fatalf must not be called.
+func (s *influxMockServer) recordQuery(r *http.Request) {
+	user, pass, ok := r.BasicAuth()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.requests = append(s.requests, influxRequest{
+		Query:    r.URL.Query(),
+		User:     user,
+		Password: pass,
+		HasAuth:  ok,
+	})
+}
+
+// queryRequests returns a copy of the recorded /query requests.
+func (s *influxMockServer) queryRequests() []influxRequest {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return slices.Clone(s.requests)
+}
+
+// handlePing must answer 204: the client library treats any other status as a
+// failed ping, and vmctl pings before querying.
+//
+// X-Influxdb-Version is set because real InfluxDB sets it. The client returns
+// the value from Ping, but vmctl discards it, so it is not asserted anywhere and
+// its content is arbitrary.
+//
+// The request is not recorded: /ping carries no db or rp to assert.
+func (s *influxMockServer) handlePing(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("X-Influxdb-Version", "1.8.0-mock")
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// influxResponse mirrors the InfluxDB 1.x JSON query response.
+type influxResponse struct {
+	Results []influxResult `json:"results"`
+}
+
+type influxResult struct {
+	StatementID int         `json:"statement_id"`
+	Series      []influxRow `json:"series,omitempty"`
+	Err         string      `json:"error,omitempty"`
+}
+
+type influxRow struct {
+	Name    string   `json:"name,omitempty"`
+	Columns []string `json:"columns"`
+	Values  [][]any  `json:"values,omitempty"`
+}
+
+// parseSelect splits a SELECT built by vmctl into its field, measurement and
+// WHERE clause. The statement shape is fixed by Series.fetchQuery:
+//
+//	select "value" from "cpu" where "host"::tag='h1' and "env"::tag=''
+//
+// Identifiers containing a double quote are not supported, which is fine
+// because no test fixture uses one.
+func parseSelect(q string) (field, measurement, where string, err error) {
+	rest, ok := strings.CutPrefix(q, `select "`)
+	if !ok {
+		return "", "", "", fmt.Errorf("cannot parse select statement: %s", q)
+	}
+	field, rest, ok = strings.Cut(rest, `" from "`)
+	if !ok {
+		return "", "", "", fmt.Errorf("cannot parse measurement in: %s", q)
+	}
+	measurement, rest, ok = strings.Cut(rest, `"`)
+	if !ok {
+		return "", "", "", fmt.Errorf("unterminated measurement in: %s", q)
+	}
+	return field, measurement, strings.TrimPrefix(rest, " where "), nil
+}
+
+// handleQuery serves the InfluxQL statements vmctl issues. Both the plain and
+// the chunked code paths of the client accept a single JSON response object,
+// so one implementation covers both.
+func (s *influxMockServer) handleQuery(w http.ResponseWriter, r *http.Request) {
+	s.recordQuery(r)
+
+	q := strings.TrimSpace(r.URL.Query().Get("q"))
+	lower := strings.ToLower(q)
+
+	// Content-Type is load-bearing: the client rejects any response that is not
+	// application/json. X-Influxdb-Version only matters for 5xx replies, where
+	// the client uses its absence to report a downstream proxy error instead of
+	// an InfluxDB one; it is set here to mirror real InfluxDB.
+	w.Header().Set("X-Influxdb-Version", "1.8.0-mock")
+	w.Header().Set("Content-Type", "application/json")
+
+	var resp influxResponse
+	switch {
+	case strings.HasPrefix(lower, "show field keys"):
+		resp = influxResponse{Results: []influxResult{{Series: s.fieldKeys()}}}
+	case strings.HasPrefix(lower, "show tag keys"):
+		resp = influxResponse{Results: []influxResult{{Series: s.tagKeys()}}}
+	case strings.HasPrefix(lower, "show series"):
+		resp = influxResponse{Results: []influxResult{{Series: s.series()}}}
+	case strings.HasPrefix(lower, "select"):
+		rows, err := s.selectRows(q)
+		if err != nil {
+			resp = influxResponse{Results: []influxResult{{Err: err.Error()}}}
+			break
+		}
+		resp = influxResponse{Results: []influxResult{{Series: rows}}}
+	default:
+		resp = influxResponse{Results: []influxResult{{Err: fmt.Sprintf("unsupported query: %s", q)}}}
+	}
+
+	_ = json.NewEncoder(w).Encode(resp)
+}
+
+// fieldKeys serves `SHOW FIELD KEYS`: one row per measurement listing every
+// field key together with its type. vmctl uses the type to skip string fields.
+//
+// Rows are built by walking the points in order, so the output is deterministic
+// without sorting.
+func (s *influxMockServer) fieldKeys() []influxRow {
+	rows := make(map[string]*influxRow)
+	var order []string
+	seen := make(map[string]struct{})
+	for _, p := range s.points {
+		row, ok := rows[p.Measurement]
+		if !ok {
+			row = &influxRow{Name: p.Measurement, Columns: []string{"fieldKey", "fieldType"}}
+			rows[p.Measurement] = row
+			order = append(order, p.Measurement)
+		}
+		key := p.Measurement + "\x00" + p.Field
+		if _, ok := seen[key]; ok {
+			continue
+		}
+		seen[key] = struct{}{}
+		row.Values = append(row.Values, []any{p.Field, p.FieldType})
+	}
+	out := make([]influxRow, 0, len(order))
+	for _, m := range order {
+		out = append(out, *rows[m])
+	}
+	return out
+}
+
+// tagKeys serves `SHOW TAG KEYS`: one row per measurement listing its tag keys.
+//
+// Tag keys come from a map, so each row is sorted to keep the output stable.
+func (s *influxMockServer) tagKeys() []influxRow {
+	rows := make(map[string]*influxRow)
+	var order []string
+	seen := make(map[string]struct{})
+	for _, p := range s.points {
+		row, ok := rows[p.Measurement]
+		if !ok {
+			row = &influxRow{Name: p.Measurement, Columns: []string{"tagKey"}}
+			rows[p.Measurement] = row
+			order = append(order, p.Measurement)
+		}
+		for k := range p.Tags {
+			key := p.Measurement + "\x00" + k
+			if _, ok := seen[key]; ok {
+				continue
+			}
+			seen[key] = struct{}{}
+			row.Values = append(row.Values, []any{k})
+		}
+	}
+	out := make([]influxRow, 0, len(order))
+	for _, m := range order {
+		row := *rows[m]
+		sort.Slice(row.Values, func(a, b int) bool {
+			return row.Values[a][0].(string) < row.Values[b][0].(string)
+		})
+		out = append(out, row)
+	}
+	return out
+}
+
+// series serves `SHOW SERIES`: a single row of series keys in the
+// `measurement,tag=value,...` form.
+func (s *influxMockServer) series() []influxRow {
+	seen := make(map[string]struct{})
+	var keys []string
+	for _, p := range s.points {
+		key := influxSeriesKey(p.Measurement, p.Tags)
+		if _, ok := seen[key]; ok {
+			continue
+		}
+		seen[key] = struct{}{}
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	row := influxRow{Columns: []string{"key"}}
+	for _, k := range keys {
+		row.Values = append(row.Values, []any{k})
+	}
+	return []influxRow{row}
+}
+
+// selectRows serves the per-series SELECT issued for every measurement/field
+// combination, returning the `time` and field columns.
+func (s *influxMockServer) selectRows(q string) ([]influxRow, error) {
+	field, measurement, where, err := parseSelect(q)
+	if err != nil {
+		return nil, err
+	}
+
+	// Tags the series must carry. An empty value means the tag must be absent,
+	// which is how vmctl addresses series that lack a tag of the measurement.
+	wantTags := make(map[string]string)
+	for _, cond := range strings.Split(where, " and ") {
+		// Conditions that are not tag comparisons - such as the time filter
+		// added by -influx-filter-time-start/-end - are not series selectors.
+		name, value, ok := strings.Cut(cond, "::tag=")
+		if !ok {
+			continue
+		}
+		wantTags[strings.Trim(name, `"`)] = strings.Trim(value, `'`)
+	}
+
+	row := influxRow{Name: measurement, Columns: []string{"time", field}}
+	for _, p := range s.points {
+		if p.Measurement != measurement || p.Field != field {
+			continue
+		}
+		if !influxTagsMatch(p.Tags, wantTags) {
+			continue
+		}
+		ts := time.Unix(p.Timestamp, 0).UTC().Format(time.RFC3339)
+		row.Values = append(row.Values, []any{ts, p.Value})
+	}
+	if len(row.Values) == 0 {
+		return nil, nil
+	}
+	return []influxRow{row}, nil
+}
+
+// influxTagsMatch reports whether the series tags satisfy the conditions of a
+// SELECT built by vmctl. Every requested tag must match exactly; a requested
+// empty value requires the tag to be absent from the series.
+func influxTagsMatch(got, want map[string]string) bool {
+	for k, v := range want {
+		if v == "" {
+			if _, ok := got[k]; ok {
+				return false
+			}
+			continue
+		}
+		if got[k] != v {
+			return false
+		}
+	}
+	// Every tag of the series must be constrained, otherwise the SELECT would
+	// address more than one series.
+	for k := range got {
+		if _, ok := want[k]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+// influxSeriesKey builds the `measurement,tag=value,...` key used by SHOW SERIES.
+//
+// The tags must be sorted: the key doubles as the deduplication key, so an
+// unstable order would report the same series more than once. tagsKey already
+// sorts, so it is reused here.
+func influxSeriesKey(measurement string, tags map[string]string) string {
+	if len(tags) == 0 {
+		return measurement
+	}
+	return measurement + "," + tagsKey(tags)
+}

apptest/tests/vmctl_influx_migration_test.go

@@ -0,0 +1,272 @@
+package tests
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/apptest"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
+)
+
+const (
+	influxTestDatabase = "testdb"
+	influxTestToken    = "my-secret-token"
+	// defaultV1CompatUser mirrors the username vmctl sends when authenticating
+	// to InfluxDB 2.x with an API token. The 1.x compatibility API requires a
+	// username but ignores its value.
+	influxV1CompatUser = "vmctl"
+)
+
+func TestSingleVmctlInfluxV2Migration(t *testing.T) {
+	fs.MustRemoveDir(t.Name())
+
+	tc := apptest.NewTestCase(t)
+	defer tc.Stop()
+
+	vmsingleDst := tc.MustStartDefaultVmsingle()
+	vmAddr := fmt.Sprintf("http://%s/", vmsingleDst.HTTPAddr())
+
+	baseTS := time.Now().Add(-2 * time.Hour).Truncate(time.Minute).Unix()
+	points := newInfluxTestPoints(baseTS)
+
+	influx := newInfluxMockServer(t, points)
+	defer influx.close()
+
+	vmctlFlags := []string{
+		`influx`,
+		`--influx-version=2`,
+		`--influx-addr=` + influx.httpAddr(),
+		`--influx-token=` + influxTestToken,
+		`--influx-database=` + influxTestDatabase,
+		`--vm-addr=` + vmAddr,
+		`--disable-progress-bar=true`,
+		`-s`,
+	}
+
+	testVmctlInfluxMigration(tc, vmsingleDst, vmctlFlags, points, baseTS)
+
+	assertInfluxRequests(t, influx, influxV1CompatUser, influxTestToken, "")
+}
+
+func TestSingleVmctlInfluxV1Migration(t *testing.T) {
+	fs.MustRemoveDir(t.Name())
+
+	tc := apptest.NewTestCase(t)
+	defer tc.Stop()
+
+	vmsingleDst := tc.MustStartDefaultVmsingle()
+	vmAddr := fmt.Sprintf("http://%s/", vmsingleDst.HTTPAddr())
+
+	baseTS := time.Now().Add(-2 * time.Hour).Truncate(time.Minute).Unix()
+	points := newInfluxTestPoints(baseTS)
+
+	influx := newInfluxMockServer(t, points)
+	defer influx.close()
+
+	vmctlFlags := []string{
+		`influx`,
+		`--influx-addr=` + influx.httpAddr(),
+		`--influx-user=user`,
+		`--influx-password=pass`,
+		`--influx-database=` + influxTestDatabase,
+		`--vm-addr=` + vmAddr,
+		`--disable-progress-bar=true`,
+		`-s`,
+	}
+
+	testVmctlInfluxMigration(tc, vmsingleDst, vmctlFlags, points, baseTS)
+	assertInfluxRequests(t, influx, "user", "pass", "autogen")
+}
+
+func TestClusterVmctlInfluxV2Migration(t *testing.T) {
+	fs.MustRemoveDir(t.Name())
+
+	tc := apptest.NewTestCase(t)
+	defer tc.Stop()
+
+	cluster := tc.MustStartDefaultCluster()
+	vmAddr := fmt.Sprintf("http://%s/", cluster.Vminsert.HTTPAddr())
+
+	baseTS := time.Now().Add(-2 * time.Hour).Truncate(time.Minute).Unix()
+	points := newInfluxTestPoints(baseTS)
+
+	influx := newInfluxMockServer(t, points)
+	defer influx.close()
+
+	vmctlFlags := []string{
+		`influx`,
+		`--influx-version=2`,
+		`--influx-addr=` + influx.httpAddr(),
+		`--influx-token=` + influxTestToken,
+		`--influx-database=` + influxTestDatabase,
+		`--vm-addr=` + vmAddr,
+		`--vm-account-id=0`,
+		`--disable-progress-bar=true`,
+		`-s`,
+	}
+
+	testVmctlInfluxMigration(tc, cluster, vmctlFlags, points, baseTS)
+	assertInfluxRequests(t, influx, influxV1CompatUser, influxTestToken, "")
+}
+
+func testVmctlInfluxMigration(
+	tc *apptest.TestCase,
+	queries apptest.PrometheusWriteQuerier,
+	vmctlFlags []string,
+	points []influxPoint,
+	baseTS int64,
+) {
+	t := tc.T()
+	t.Helper()
+
+	queryStart := time.Unix(baseTS-3600, 0).UTC().Format(time.RFC3339)
+	queryEnd := time.Unix(baseTS+7200, 0).UTC().Format(time.RFC3339)
+
+	cmpOpt := cmpopts.IgnoreFields(apptest.PrometheusAPIV1QueryResponse{}, "Status", "Data.ResultType")
+
+	// Nothing is stored before the migration runs.
+	got := queries.PrometheusAPIV1Query(t, `{__name__=~".*"}`, apptest.QueryOpts{
+		Step: "5m",
+		Time: queryStart,
+	})
+	want := apptest.NewPrometheusAPIV1QueryResponse(t, `{"data":{"result":[]}}`)
+	if diff := cmp.Diff(want, got, cmpOpt); diff != "" {
+		t.Errorf("unexpected response before migration (-want, +got):\n%s", diff)
+	}
+
+	tc.MustStartVmctl("vmctl", vmctlFlags)
+	queries.ForceFlush(t)
+
+	tc.Assert(&apptest.AssertOptions{
+		Retries: 300,
+		Msg:     `unexpected metrics migrated from influx`,
+		Got: func() any {
+			r := queries.PrometheusAPIV1Export(t, `{__name__!=""}`, apptest.QueryOpts{
+				Start: queryStart,
+				End:   queryEnd,
+			})
+			r.Sort()
+			return r.Data.Result
+		},
+		Want: buildExpectedInfluxResult(t, points),
+		CmpOpts: []cmp.Option{
+			cmpopts.IgnoreFields(apptest.PrometheusAPIV1QueryResponse{}, "Status", "Data.ResultType"),
+		},
+	})
+}
+
+// buildExpectedInfluxResult derives the expected VictoriaMetrics contents from
+// the same points the mock server serves: metric name is
+// `<measurement>_<field>`, tags become labels, the source database is added as
+// the `db` label, booleans become 1/0 and string fields are skipped.
+func buildExpectedInfluxResult(t *testing.T, points []influxPoint) []*apptest.QueryResult {
+	t.Helper()
+
+	grouped := map[string]*apptest.QueryResult{}
+	for _, p := range points {
+		if p.FieldType == "string" {
+			continue
+		}
+		name := fmt.Sprintf("%s_%s", p.Measurement, p.Field)
+		metric := map[string]string{
+			"__name__": name,
+			"db":       influxTestDatabase,
+		}
+		for k, v := range p.Tags {
+			metric[k] = v
+		}
+		key := tagsKey(metric)
+		if _, ok := grouped[key]; !ok {
+			grouped[key] = &apptest.QueryResult{Metric: metric}
+		}
+		grouped[key].Samples = append(grouped[key].Samples, &apptest.Sample{
+			Timestamp: p.Timestamp * 1000,
+			Value:     influxExpectedValue(t, p.Value),
+		})
+	}
+	out := make([]*apptest.QueryResult, 0, len(grouped))
+	for _, v := range grouped {
+		out = append(out, v)
+	}
+	resp := apptest.PrometheusAPIV1QueryResponse{
+		Data: &apptest.QueryData{Result: out},
+	}
+	resp.Sort()
+	return resp.Data.Result
+}
+
+// influxExpectedValue converts a field value the way vmctl does: numbers pass
+// through and booleans become 1/0.
+func influxExpectedValue(t *testing.T, v any) float64 {
+	t.Helper()
+
+	switch val := v.(type) {
+	case float64:
+		return val
+	case int64:
+		return float64(val)
+	case bool:
+		if val {
+			return 1
+		}
+		return 0
+	default:
+		t.Fatalf("unexpected field value type %T in test fixture; only float64, int64 and bool are supported", v)
+		return 0
+	}
+}
+
+// assertInfluxRequests verifies how vmctl authenticated to InfluxDB and which
+// database and retention policy it requested.
+func assertInfluxRequests(t *testing.T, s *influxMockServer, wantUser, wantPass, wantRP string) {
+	t.Helper()
+
+	reqs := s.queryRequests()
+	if len(reqs) == 0 {
+		t.Fatalf("vmctl issued no /query requests")
+	}
+	for _, r := range reqs {
+		if !r.HasAuth {
+			t.Fatalf("no basic auth on %s", r.Query)
+		}
+		if r.User != wantUser || r.Password != wantPass {
+			t.Fatalf("unexpected credentials for q=%q\ngot\n(%q, %q)\nwant\n(%q, %q)",
+				r.Query.Get("q"), r.User, r.Password, wantUser, wantPass)
+		}
+		if rp := r.Query.Get("rp"); rp != wantRP {
+			t.Fatalf("unexpected rp for q=%q\ngot\n%q\nwant\n%q", r.Query.Get("q"), rp, wantRP)
+		}
+		if db := r.Query.Get("db"); db != influxTestDatabase {
+			t.Fatalf("unexpected db for q=%q\ngot\n%q\nwant\n%q", r.Query.Get("q"), db, influxTestDatabase)
+		}
+	}
+}
+
+// newInfluxTestPoints returns points covering the cases that matter for the
+// migration: numeric field types, a boolean, a string field which must be
+// skipped, a series missing one of the measurement tags, and a measurement name
+// containing special characters.
+func newInfluxTestPoints(baseTS int64) []influxPoint {
+	var points []influxPoint
+	for i := range 10 {
+		ts := baseTS + int64(i*60)
+
+		// Two series of `cpu`: the second one has no `env` tag, which exercises
+		// the empty-tag conditions vmctl adds to its SELECT statements.
+		points = append(points,
+			influxPoint{"cpu", map[string]string{"host": "h1", "env": "prod"}, "value", "float", ts, float64(i) + 0.5},
+			influxPoint{"cpu", map[string]string{"host": "h1", "env": "prod"}, "count", "integer", ts, int64(i)},
+			influxPoint{"cpu", map[string]string{"host": "h1", "env": "prod"}, "flag", "boolean", ts, i%2 == 0},
+			influxPoint{"cpu", map[string]string{"host": "h1", "env": "prod"}, "note", "string", ts, "ignored"},
+			influxPoint{"cpu", map[string]string{"host": "h2"}, "value", "float", ts, float64(i) - 2.25},
+			influxPoint{"cpu", map[string]string{"host": "h2"}, "note", "string", ts, "ignored"},
+			// Special characters in a measurement name, cf. issue #10892.
+			influxPoint{"user_percent.mem.zwickel+", map[string]string{"host": "h1"}, "value", "float", ts, float64(i * 3)},
+		)
+	}
+	return points
+}