feat(cmd/influxd): export 1.x continuous queries as part of upgrade

CHANGELOG.md

@@ -2,7 +2,8 @@
 
 ### Features
 
-1. [20036](https://github.com/influxdata/influxdb/pull/): warn if V1 users are upgraded, but V1 auth wasn't enabled
+1. [20036](https://github.com/influxdata/influxdb/pull/20036): Warn if V1 users are upgraded, but V1 auth wasn't enabled
+1. [20039](https://github.com/influxdata/influxdb/pull/20039): Export 1.x CQs as part of `influxd upgrade`
 
 ### Bug Fixes
 

cmd/influxd/upgrade/database.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"fmt"
 	"github.com/dustin/go-humanize"
+	"os"
 	"path/filepath"
+	"strings"
 
 	"github.com/influxdata/influxdb/v2"
 	"github.com/influxdata/influxdb/v2/pkg/fs"
@@ -55,10 +57,18 @@ func upgradeDatabases(ctx context.Context, v1 *influxDBv1, v2 *influxDBv2, v1opt
 	if size > diskInfo.Free {
 		return nil, fmt.Errorf("not enough space on target disk of %s: need %d, available %d ", v2dir, size, diskInfo.Free)
 	}
+
+	cqFile, err := os.OpenFile(v2opts.cqPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		return nil, fmt.Errorf("error opening file for CQ export %s: %w", v2opts.cqPath, err)
+	}
+	defer cqFile.Close()
+
 	log.Info("Upgrading databases")
 	// read each database / retention policy from v1.meta and create bucket db-name/rp-name
 	// create database in v2.meta
 	// copy shard info from v1.meta
+	// export any continuous queries
 	for _, db := range v1.meta.Databases() {
 		if db.Name == "_internal" {
 			if options.verbose {
@@ -174,6 +184,40 @@ func upgradeDatabases(ctx context.Context, v1 *influxDBv1, v2 *influxDBv2, v1opt
 				log.Warn("Empty retention policy")
 			}
 		}
+
+		// Output CQs in the same format as SHOW CONTINUOUS QUERIES
+		_, err := cqFile.WriteString(fmt.Sprintf("name: %s\n", db.Name))
+		if err != nil {
+			return nil, err
+		}
+		maxNameLen := 4 // 4 == len("name"), the column header
+		for _, cq := range db.ContinuousQueries {
+			if len(cq.Name) > maxNameLen {
+				maxNameLen = len(cq.Name)
+			}
+		}
+
+		headerPadding := maxNameLen - 4 + 1
+		_, err = cqFile.WriteString(fmt.Sprintf("name%[1]squery\n----%[1]s-----\n", strings.Repeat(" ", headerPadding)))
+		if err != nil {
+			return nil, err
+		}
+
+		for _, cq := range db.ContinuousQueries {
+			if options.verbose {
+				log.Info("Exporting CQ", zap.String("db", db.Name), zap.String("cq_name", cq.Name))
+			}
+			padding := maxNameLen - len(cq.Name) + 1
+
+			_, err := cqFile.WriteString(fmt.Sprintf("%s%s%s\n", cq.Name, strings.Repeat(" ", padding), cq.Query))
+			if err != nil {
+				return nil, fmt.Errorf("error exporting continuous query %s from DB %s: %w", cq.Name, db.Name, err)
+			}
+		}
+		_, err = cqFile.WriteString("\n")
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	return db2BucketIds, nil

cmd/influxd/upgrade/database_test.go

@@ -38,6 +38,7 @@ func TestUpgradeRealDB(t *testing.T) {
 
 	boltPath := filepath.Join(tl.Path, bolt.DefaultFilename)
 	enginePath := filepath.Join(tl.Path, "engine")
+	cqPath := filepath.Join(tl.Path, "cq.txt")
 
 	v1opts := &optionsV1{dbDir: tmpdir + "/v1db"}
 	v1opts.populateDirs()
@@ -48,6 +49,7 @@ func TestUpgradeRealDB(t *testing.T) {
 	v2opts := &optionsV2{
 		boltPath:   boltPath,
 		enginePath: enginePath,
+		cqPath:     cqPath,
 		userName:   "my-user",
 		password:   "my-password",
 		orgName:    "my-org",
@@ -186,6 +188,14 @@ func TestUpgradeRealDB(t *testing.T) {
 
 	respBody = mustRunQuery(t, tl, "mydb", `select count(line) from mydb."1week".log`)
 	assert.Contains(t, respBody, `["1970-01-01T00:00:00Z",1]`)
+
+	cqBytes, err := ioutil.ReadFile(cqPath)
+	require.NoError(t, err)
+	cqs := string(cqBytes)
+
+	assert.Contains(t, cqs, "CREATE CONTINUOUS QUERY other_cq ON test BEGIN SELECT mean(foo) INTO test.autogen.foo FROM empty.autogen.foo GROUP BY time(1h) END")
+	assert.Contains(t, cqs, "CREATE CONTINUOUS QUERY cq_3 ON test BEGIN SELECT mean(bar) INTO test.autogen.bar FROM test.autogen.foo GROUP BY time(1m) END")
+	assert.Contains(t, cqs, "CREATE CONTINUOUS QUERY cq ON empty BEGIN SELECT mean(example) INTO empty.autogen.mean FROM empty.autogen.raw GROUP BY time(1h) END")
 }
 
 func mustRunQuery(t *testing.T, tl *launcher.TestLauncher, db, rawQ string) string {

cmd/influxd/upgrade/upgrade.go

@@ -90,6 +90,7 @@ type optionsV2 struct {
 	boltPath       string
 	cliConfigsPath string
 	enginePath     string
+	cqPath         string
 	userName       string
 	password       string
 	orgName        string
@@ -139,6 +140,7 @@ func NewCommand(v *viper.Viper) *cobra.Command {
       1. Reads the 1.x config file and creates a 2.x config file with matching options. Unsupported 1.x options are reported.
       2. Copies 1.x database files.
       3. Creates influx CLI configurations.
+      4. Exports any 1.x continuous queries to disk.
 
     If the config file is not available, 1.x db folder (--v1-dir options) is taken as an input.
     Target 2.x database dir is specified by the --engine-path option. If changed, the bolt path should be changed as well.
@@ -182,6 +184,12 @@ func NewCommand(v *viper.Viper) *cobra.Command {
 			Desc:    "path for persistent engine files",
 			Short:   'e',
 		},
+		{
+			DestP:   &options.target.cqPath,
+			Flag:    "continuous-query-export-path",
+			Default: filepath.Join(homeOrAnyDir(), "continuous_queries.txt"),
+			Desc:    "path for exported 1.x continuous queries",
+		},
 		{
 			DestP:    &options.target.userName,
 			Flag:     "username",
@@ -469,6 +477,10 @@ func validatePaths(sourceOpts *optionsV1, targetOpts *optionsV2) error {
 		return fmt.Errorf("file present at target path for 2.x CLI configs '%s'", targetOpts.cliConfigsPath)
 	}
 
+	if _, err = os.Stat(targetOpts.cqPath); err == nil {
+		return fmt.Errorf("file present at target path for exported continuous queries '%s'", targetOpts.cqPath)
+	}
+
 	return nil
 }