feat(cmd/influx): allow setting shard-group durations for buckets via…

… API and CLI (#20911)



Co-authored-by: Alexander Savinykh <[email protected]>

CHANGELOG.md

@@ -23,6 +23,7 @@ or `/query` HTTP endpoints.
 1. [20827](https://github.com/influxdata/influxdb/pull/20827): Add `--pprof-disabled` option to `influxd` to disable exposing profiling information over HTTP.
 1. [20827](https://github.com/influxdata/influxdb/pull/20827): Add `/debug/pprof/all` HTTP endpoint to gather all profiles at once.
 1. [20827](https://github.com/influxdata/influxdb/pull/20827): Upgrade `http.pprof-enabled` config in `influxd upgrade`.
+1. [20911](https://github.com/influxdata/influxdb/pull/20911): Add support for explicitly setting shard-group durations on buckets. Thanks @hinst!
 
 ### Bug Fixes
 

bucket.go

@@ -36,6 +36,7 @@ type Bucket struct {
 	Description         string        `json:"description"`
 	RetentionPolicyName string        `json:"rp,omitempty"` // This to support v1 sources
 	RetentionPeriod     time.Duration `json:"retentionPeriod"`
+	ShardGroupDuration  time.Duration `json:"shardGroupDuration"`
 	CRUDLog
 }
 
@@ -102,9 +103,10 @@ type BucketService interface {
 // BucketUpdate represents updates to a bucket.
 // Only fields which are set are updated.
 type BucketUpdate struct {
-	Name            *string        `json:"name,omitempty"`
-	Description     *string        `json:"description,omitempty"`
-	RetentionPeriod *time.Duration `json:"retentionPeriod,omitempty"`
+	Name               *string
+	Description        *string
+	RetentionPeriod    *time.Duration
+	ShardGroupDuration *time.Duration
 }
 
 // BucketFilter represents a set of filter that restrict the returned results.

cmd/influx/bucket.go

@@ -23,13 +23,14 @@ type cmdBucketBuilder struct {
 
 	svcFn bucketSVCsFn
 
-	id          string
-	hideHeaders bool
-	json        bool
-	name        string
-	description string
-	org         organization
-	retention   string
+	id                 string
+	hideHeaders        bool
+	json               bool
+	name               string
+	description        string
+	org                organization
+	retention          string
+	shardGroupDuration string
 }
 
 func newCmdBucketBuilder(svcsFn bucketSVCsFn, f *globalFlags, opts genericCLIOpts) *cmdBucketBuilder {
@@ -73,6 +74,8 @@ func (b *cmdBucketBuilder) cmdCreate() *cobra.Command {
 
 	cmd.Flags().StringVarP(&b.description, "description", "d", "", "Description of bucket that will be created")
 	cmd.Flags().StringVarP(&b.retention, "retention", "r", "", "Duration bucket will retain data. 0 is infinite. Default is 0.")
+	cmd.Flags().StringVarP(&b.shardGroupDuration, "shard-group-duration", "", "",
+		"Shard group duration used internally by the storage engine. Not supported by InfluxDB Cloud.")
 	b.org.register(b.viper, cmd, false)
 	b.registerPrintFlags(cmd)
 
@@ -94,10 +97,16 @@ func (b *cmdBucketBuilder) cmdCreateRunEFn(*cobra.Command, []string) error {
 		return err
 	}
 
+	shardGroupDuration, err := internal.RawDurationToTimeDuration(b.shardGroupDuration)
+	if err != nil {
+		return err
+	}
+
 	bkt := &influxdb.Bucket{
-		Name:            b.name,
-		Description:     b.description,
-		RetentionPeriod: dur,
+		Name:               b.name,
+		Description:        b.description,
+		RetentionPeriod:    dur,
+		ShardGroupDuration: shardGroupDuration,
 	}
 	bkt.OrgID, err = b.org.getID(orgSVC)
 	if err != nil {
@@ -241,16 +250,18 @@ func (b *cmdBucketBuilder) cmdUpdate() *cobra.Command {
 			Flag:   "name",
 			Short:  'n',
 			EnvVar: "BUCKET_NAME",
-			Desc:   "New bucket name",
+			Desc:   "New name to set on the bucket",
 		},
 	}
 	opts.mustRegister(b.viper, cmd)
 
 	b.registerPrintFlags(cmd)
 	cmd.Flags().StringVarP(&b.id, "id", "i", "", "The bucket ID (required)")
-	cmd.Flags().StringVarP(&b.description, "description", "d", "", "Description of bucket that will be created")
+	cmd.Flags().StringVarP(&b.description, "description", "d", "", "New description to set on the bucket")
 	cmd.MarkFlagRequired("id")
-	cmd.Flags().StringVarP(&b.retention, "retention", "r", "", "Duration bucket will retain data. 0 is infinite. Default is 0.")
+	cmd.Flags().StringVarP(&b.retention, "retention", "r", "", "New retention duration to set on the bucket. 0 is infinite.")
+	cmd.Flags().StringVarP(&b.shardGroupDuration, "shard-group-duration", "", "",
+		"New shard group duration to set on the bucket. 0 will tell the server to pick a value. Not supported by InfluxDB Cloud.")
 
 	return cmd
 }
@@ -274,14 +285,22 @@ func (b *cmdBucketBuilder) cmdUpdateRunEFn(cmd *cobra.Command, args []string) er
 		update.Description = &b.description
 	}
 
-	dur, err := internal.RawDurationToTimeDuration(b.retention)
-	if err != nil {
-		return err
-	}
-	if dur != 0 {
+	if b.retention != "" {
+		dur, err := internal.RawDurationToTimeDuration(b.retention)
+		if err != nil {
+			return err
+		}
 		update.RetentionPeriod = &dur
 	}
 
+	if b.shardGroupDuration != "" {
+		sgDur, err := internal.RawDurationToTimeDuration(b.shardGroupDuration)
+		if err != nil {
+			return err
+		}
+		update.ShardGroupDuration = &sgDur
+	}
+
 	bkt, err := bktSVC.UpdateBucket(context.Background(), id, update)
 	if err != nil {
 		return fmt.Errorf("failed to update bucket: %v", err)
@@ -320,7 +339,7 @@ func (b *cmdBucketBuilder) printBuckets(printOpt bucketPrintOpt) error {
 
 	w.HideHeaders(b.hideHeaders)
 
-	headers := []string{"ID", "Name", "Retention", "Organization ID"}
+	headers := []string{"ID", "Name", "Retention", "Shard group duration", "Organization ID"}
 	if printOpt.deleted {
 		headers = append(headers, "Deleted")
 	}
@@ -331,11 +350,23 @@ func (b *cmdBucketBuilder) printBuckets(printOpt bucketPrintOpt) error {
 	}
 
 	for _, bkt := range printOpt.buckets {
+		rp := bkt.RetentionPeriod.String()
+		if bkt.RetentionPeriod == influxdb.InfiniteRetention {
+			rp = "infinite"
+		}
+		sgDur := bkt.ShardGroupDuration.String()
+		// ShardGroupDuration will be zero if listing buckets from InfluxDB Cloud.
+		// Show something more useful here in that case.
+		if bkt.ShardGroupDuration == 0 {
+			sgDur = "n/a"
+		}
+
 		m := map[string]interface{}{
-			"ID":              bkt.ID.String(),
-			"Name":            bkt.Name,
-			"Retention":       bkt.RetentionPeriod,
-			"Organization ID": bkt.OrgID.String(),
+			"ID":                   bkt.ID.String(),
+			"Name":                 bkt.Name,
+			"Retention":            rp,
+			"Shard group duration": sgDur,
+			"Organization ID":      bkt.OrgID.String(),
 		}
 		if printOpt.deleted {
 			m["Deleted"] = true

cmd/influx/bucket_test.go

@@ -90,6 +90,21 @@ func TestCmdBucket(t *testing.T) {
 					OrgID:           orgID,
 				},
 			},
+			{
+				name: "with explicit shard-group duration",
+				flags: []string{
+					"-r=1h",
+					"--shard-group-duration=1m",
+					"-o=org name",
+					"-n=new name",
+				},
+				expectedBucket: influxdb.Bucket{
+					Name:               "new name",
+					RetentionPeriod:    time.Hour,
+					ShardGroupDuration: time.Minute,
+					OrgID:              orgID,
+				},
+			},
 		}
 
 		cmdFn := func(expectedBkt influxdb.Bucket) func(*globalFlags, genericCLIOpts) *cobra.Command {
@@ -408,6 +423,16 @@ func TestCmdBucket(t *testing.T) {
 					RetentionPeriod: durPtr(time.Minute),
 				},
 			},
+			{
+				name: "shard-group duration",
+				flags: []string{
+					"-i=" + influxdb.ID(3).String(),
+					"--shard-group-duration=1m",
+				},
+				expected: influxdb.BucketUpdate{
+					ShardGroupDuration: durPtr(time.Minute),
+				},
+			},
 		}
 
 		cmdFn := func(expectedUpdate influxdb.BucketUpdate) func(*globalFlags, genericCLIOpts) *cobra.Command {

cmd/influxd/launcher/engine.go

@@ -127,8 +127,8 @@ func (t *TemporaryEngine) CreateBucket(ctx context.Context, b *influxdb.Bucket)
 	return t.engine.CreateBucket(ctx, b)
 }
 
-func (t *TemporaryEngine) UpdateBucketRetentionPeriod(ctx context.Context, bucketID influxdb.ID, d time.Duration) error {
-	return t.engine.UpdateBucketRetentionPeriod(ctx, bucketID, d)
+func (t *TemporaryEngine) UpdateBucketRetentionPolicy(ctx context.Context, bucketID influxdb.ID, upd *influxdb.BucketUpdate) error {
+	return t.engine.UpdateBucketRetentionPolicy(ctx, bucketID, upd)
 }
 
 // DeleteBucket deletes a bucket from the time-series data.

cmd/influxd/launcher/storage_test.go

@@ -8,11 +8,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/dustin/go-humanize"
 	"github.com/google/go-cmp/cmp"
 	"github.com/influxdata/influxdb/v2"
 	"github.com/influxdata/influxdb/v2/cmd/influxd/launcher"
 	"github.com/influxdata/influxdb/v2/http"
-	"github.com/influxdata/influxdb/v2/pkg/testing/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -200,15 +200,130 @@ func TestLauncher_DeleteWithPredicate(t *testing.T) {
 }
 
 func TestLauncher_UpdateRetentionPolicy(t *testing.T) {
-	l := launcher.RunAndSetupNewLauncherOrFail(ctx, t)
-	defer l.ShutdownOrFail(t, ctx)
+	durPtr := func(d time.Duration) *time.Duration {
+		return &d
+	}
 
-	bucket, err := l.BucketService(t).FindBucket(ctx, influxdb.BucketFilter{ID: &l.Bucket.ID})
-	require.NoError(t, err)
-	require.NotNil(t, bucket)
+	testCases := []struct {
+		name            string
+		initRp          time.Duration
+		initSgd         time.Duration
+		derivedSgd      *time.Duration
+		newRp           *time.Duration
+		newSgd          *time.Duration
+		expectInitErr   bool
+		expectUpdateErr bool
+	}{
+		{
+			name:       "infinite to 1w",
+			derivedSgd: durPtr(humanize.Week),
+			newRp:      durPtr(humanize.Week),
+		},
+		{
+			name:       "1w to 1d",
+			initRp:     humanize.Week,
+			derivedSgd: durPtr(humanize.Day),
+			newRp:      durPtr(humanize.Day),
+		},
+		{
+			name:       "1d to 1h",
+			initRp:     humanize.Day,
+			derivedSgd: durPtr(time.Hour),
+			newRp:      durPtr(time.Hour),
+		},
+		{
+			name:       "infinite, update shard duration",
+			initSgd:    humanize.Month,
+			derivedSgd: durPtr(humanize.Month),
+			newSgd:     durPtr(humanize.Week),
+		},
+		{
+			name:    "1w, update shard duration",
+			initRp:  humanize.Week,
+			initSgd: humanize.Week,
+			newSgd:  durPtr(time.Hour),
+		},
+		{
+			name:    "1d, update shard duration",
+			initRp:  humanize.Day,
+			initSgd: 3 * time.Hour,
+			newSgd:  durPtr(1*time.Hour + 30*time.Minute),
+		},
+		{
+			name:       "infinite, update both retention and shard duration",
+			derivedSgd: durPtr(humanize.Week),
+			newRp:      durPtr(time.Hour),
+			newSgd:     durPtr(time.Hour),
+		},
+		{
+			name:          "init shard duration larger than RP",
+			initRp:        time.Hour,
+			initSgd:       humanize.Day,
+			expectInitErr: true,
+		},
+		{
+			name:            "updated shard duration larger than RP",
+			initRp:          humanize.Day,
+			initSgd:         time.Hour,
+			newSgd:          durPtr(humanize.Week),
+			expectUpdateErr: true,
+		},
+	}
 
-	newRetentionPeriod := 1 * time.Hour
-	bucket, err = l.BucketService(t).UpdateBucket(ctx, bucket.ID, influxdb.BucketUpdate{RetentionPeriod: &newRetentionPeriod})
-	require.NoError(t, err)
-	assert.Equal(t, bucket.RetentionPeriod, newRetentionPeriod)
+	for _, tc := range testCases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			l := launcher.RunAndSetupNewLauncherOrFail(ctx, t)
+			defer l.ShutdownOrFail(t, ctx)
+			bucketService := l.BucketService(t)
+
+			bucket := &influxdb.Bucket{
+				OrgID:              l.Org.ID,
+				RetentionPeriod:    tc.initRp,
+				ShardGroupDuration: tc.initSgd,
+			}
+			err := bucketService.CreateBucket(ctx, bucket)
+			if tc.expectInitErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			defer bucketService.DeleteBucket(ctx, bucket.ID)
+
+			bucket, err = bucketService.FindBucketByID(ctx, bucket.ID)
+			require.NoError(t, err)
+
+			expectedSgd := tc.initSgd
+			if tc.derivedSgd != nil {
+				expectedSgd = *tc.derivedSgd
+			}
+			require.Equal(t, tc.initRp, bucket.RetentionPeriod)
+			require.Equal(t, expectedSgd, bucket.ShardGroupDuration)
+
+			bucket, err = bucketService.UpdateBucket(ctx, bucket.ID, influxdb.BucketUpdate{
+				RetentionPeriod:    tc.newRp,
+				ShardGroupDuration: tc.newSgd,
+			})
+			if tc.expectUpdateErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+
+			bucket, err = bucketService.FindBucketByID(ctx, bucket.ID)
+			require.NoError(t, err)
+
+			expectedRp := tc.initRp
+			if tc.newRp != nil {
+				expectedRp = *tc.newRp
+			}
+			if tc.newSgd != nil {
+				expectedSgd = *tc.newSgd
+			}
+			require.Equal(t, expectedRp, bucket.RetentionPeriod)
+			require.Equal(t, expectedSgd, bucket.ShardGroupDuration)
+		})
+	}
 }