Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #7449 from influxdata/cjl-7488-fix-rp-defaults 

fix retention policy creation inconsistencies
pull/7597/head
Cory LaNou 2016-11-07 08:20:34 -06:00 committed by GitHub
parent 4b85371a71 6e290040bb
commit 40f626d20f
16 changed files with 220 additions and 339 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@ -76,6 +76,7 @@ The query language has been extended with a few new features:
- [#7526](https://github.com/influxdata/influxdb/issues/7526): Truncate the version string when linking to the documentation.
- [#7548](https://github.com/influxdata/influxdb/issues/7548): Fix output duration units for SHOW QUERIES.
- [#7564](https://github.com/influxdata/influxdb/issues/7564): Fix incorrect grouping when multiple aggregates are used with sparse data.
- [#7448](https://github.com/influxdata/influxdb/pull/7448): Fix Retention Policy Inconsistencies
## v1.0.2 [2016-10-05]

5
cmd/influxd/run/backup_restore_test.go
View File

@ -34,10 +34,7 @@ func TestServer_BackupAndRestore(t *testing.T) {
s := OpenServer(config)
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy(db, newRetentionPolicySpec(rp, 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy(db, rp); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy(db, newRetentionPolicySpec(rp, 1, 0), true); err != nil {
t.Fatal(err)
}

15
cmd/influxd/run/server_helpers_test.go
View File

@ -78,10 +78,7 @@ func OpenServerWithVersion(c *run.Config, version string) *Server {
// OpenDefaultServer opens a test server with a default database & retention policy.
func OpenDefaultServer(c *run.Config) *Server {
s := OpenServer(c)
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
panic(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
panic(err)
}
return s
@ -111,10 +108,10 @@ func (s *Server) URL() string {
}
// CreateDatabaseAndRetentionPolicy will create the database and retention policy.
func (s *Server) CreateDatabaseAndRetentionPolicy(db string, rp *meta.RetentionPolicySpec) error {
func (s *Server) CreateDatabaseAndRetentionPolicy(db string, rp *meta.RetentionPolicySpec, makeDefault bool) error {
if _, err := s.MetaClient.CreateDatabase(db); err != nil {
return err
} else if _, err := s.MetaClient.CreateRetentionPolicy(db, rp); err != nil {
} else if _, err := s.MetaClient.CreateRetentionPolicy(db, rp, makeDefault); err != nil {
return err
}
return nil
@ -455,13 +452,9 @@ func writeTestData(s *Server, t *Test) error {
w.rp = t.retentionPolicy()
}
if err := s.CreateDatabaseAndRetentionPolicy(w.db, newRetentionPolicySpec(w.rp, 1, 0)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy(w.db, newRetentionPolicySpec(w.rp, 1, 0), true); err != nil {
return err
}
if err := s.MetaClient.SetDefaultRetentionPolicy(w.db, w.rp); err != nil {
return err
}
if res, err := s.Write(w.db, w.rp, w.data, t.params); err != nil {
return fmt.Errorf("write #%d: %s", i, err)
} else if t.exp != res {

6
cmd/influxd/run/server_suite_test.go
View File

@ -417,6 +417,12 @@ func init() {
exp: `{"results":[{}]}`,
once: true,
},
&Query{
name: "create retention policy with default on",
command: `CREATE RETENTION POLICY rp3 ON db0 DURATION 1h REPLICATION 1 SHARD DURATION 30m DEFAULT`,
exp: `{"results":[{"error":"retention policy conflicts with an existing policy"}]}`,
once: true,
},
&Query{
name: "show retention policy should show both with custom shard",
command: `SHOW RETENTION POLICIES ON db0`,

211
cmd/influxd/run/server_test.go
View File

@ -83,10 +83,7 @@ func TestServer_Query_DropAndRecreateDatabase(t *testing.T) {
test := tests.load(t, "drop_and_recreate_database")
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy(test.database(), test.retentionPolicy()); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0), true); err != nil {
t.Fatal(err)
}
@ -115,13 +112,10 @@ func TestServer_Query_DropDatabaseIsolated(t *testing.T) {
test := tests.load(t, "drop_database_isolated")
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0), true); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy(test.database(), test.retentionPolicy()); err != nil {
t.Fatal(err)
}
if err := s.CreateDatabaseAndRetentionPolicy("db1", newRetentionPolicySpec("rp1", 1, 0)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db1", newRetentionPolicySpec("rp1", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -150,10 +144,7 @@ func TestServer_Query_DeleteSeries(t *testing.T) {
test := tests.load(t, "delete_series")
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy(test.database(), test.retentionPolicy()); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0), true); err != nil {
t.Fatal(err)
}
@ -182,10 +173,7 @@ func TestServer_Query_DropAndRecreateSeries(t *testing.T) {
test := tests.load(t, "drop_and_recreate_series")
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy(test.database(), test.retentionPolicy()); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0), true); err != nil {
t.Fatal(err)
}
@ -234,10 +222,7 @@ func TestServer_Query_DropSeriesFromRegex(t *testing.T) {
test := tests.load(t, "drop_series_from_regex")
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy(test.database(), test.retentionPolicy()); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy(test.database(), newRetentionPolicySpec(test.retentionPolicy(), 1, 0), true); err != nil {
t.Fatal(err)
}
@ -408,7 +393,7 @@ func TestServer_Write_LineProtocol_Float(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour), true); err != nil {
t.Fatal(err)
}
@ -433,7 +418,7 @@ func TestServer_Write_LineProtocol_Bool(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour), true); err != nil {
t.Fatal(err)
}
@ -458,7 +443,7 @@ func TestServer_Write_LineProtocol_String(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour), true); err != nil {
t.Fatal(err)
}
@ -483,7 +468,7 @@ func TestServer_Write_LineProtocol_Integer(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour), true); err != nil {
t.Fatal(err)
}
@ -509,7 +494,7 @@ func TestServer_Write_LineProtocol_Partial(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 1*time.Hour), true); err != nil {
t.Fatal(err)
}
@ -2762,7 +2747,7 @@ func TestServer_Query_MergeMany(t *testing.T) {
defer s.Close()
// set infinite retention policy as we are inserting data in the past and don't want retention policy enforcement to make this test racy
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -2822,7 +2807,7 @@ func TestServer_Query_SLimitAndSOffset(t *testing.T) {
defer s.Close()
// set infinite retention policy as we are inserting data in the past and don't want retention policy enforcement to make this test racy
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -2878,10 +2863,7 @@ func TestServer_Query_Regex(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -3857,10 +3839,7 @@ func TestServer_Query_AggregateSelectors(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -4189,10 +4168,7 @@ func TestServer_Query_TopInt(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -4356,10 +4332,7 @@ func TestServer_Query_Aggregates_IdenticalTime(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -4424,10 +4397,7 @@ func TestServer_Query_GroupByTimeCutoffs(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -4506,10 +4476,7 @@ func TestServer_Write_Precision(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -4623,10 +4590,7 @@ func TestServer_Query_Wildcards(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -4741,10 +4705,7 @@ func TestServer_Query_WildcardExpansion(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -4821,10 +4782,7 @@ func TestServer_Query_AcrossShardsAndFields(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -4895,10 +4853,7 @@ func TestServer_Query_Where_Fields(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5117,10 +5072,7 @@ func TestServer_Query_Where_With_Tags(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5205,10 +5157,7 @@ func TestServer_Query_With_EmptyTags(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5302,10 +5251,7 @@ func TestServer_Query_LimitAndOffset(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5419,10 +5365,7 @@ func TestServer_Query_Fill(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5518,10 +5461,7 @@ func TestServer_Query_Chunk(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5570,16 +5510,10 @@ func TestServer_Query_DropAndRecreateMeasurement(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
t.Fatal(err)
}
if err := s.CreateDatabaseAndRetentionPolicy("db1", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db1", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db1", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5736,10 +5670,7 @@ func TestServer_Query_ShowQueries_Future(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5808,10 +5739,7 @@ func TestServer_Query_ShowSeries(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5910,10 +5838,7 @@ func TestServer_Query_ShowStats(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -5954,10 +5879,7 @@ func TestServer_Query_ShowMeasurements(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6044,10 +5966,7 @@ func TestServer_Query_ShowTagKeys(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6194,10 +6113,7 @@ func TestServer_Query_ShowFieldKeys(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6260,10 +6176,7 @@ func TestServer_ContinuousQuery(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6386,10 +6299,7 @@ func TestServer_ContinuousQuery_Deadlock(t *testing.T) {
s.Server = nil
}()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6459,10 +6369,7 @@ func TestServer_Query_EvilIdentifiers(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6503,10 +6410,7 @@ func TestServer_Query_OrderByTime(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6565,10 +6469,7 @@ func TestServer_Query_FieldWithMultiplePeriods(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6619,10 +6520,7 @@ func TestServer_Query_FieldWithMultiplePeriodsMeasurementPrefixMatch(t *testing.
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6673,10 +6571,7 @@ func TestServer_Query_IntoTarget(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6732,10 +6627,7 @@ func TestServer_Query_IntoTarget_Sparse(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6793,10 +6685,7 @@ func TestServer_Query_DuplicateMeasurements(t *testing.T) {
defer s.Close()
// Create a second database.
if err := s.CreateDatabaseAndRetentionPolicy("db1", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db1", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db1", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -6920,10 +6809,7 @@ func TestServer_WhereTimeInclusive(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}
@ -7030,10 +6916,7 @@ func TestServer_Query_ImplicitEndTime(t *testing.T) {
s := OpenServer(NewConfig())
defer s.Close()
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0)); err != nil {
t.Fatal(err)
}
if err := s.MetaClient.SetDefaultRetentionPolicy("db0", "rp0"); err != nil {
if err := s.CreateDatabaseAndRetentionPolicy("db0", newRetentionPolicySpec("rp0", 1, 0), true); err != nil {
t.Fatal(err)
}

5
coordinator/meta_client.go
View File

@ -12,7 +12,7 @@ type MetaClient interface {
CreateContinuousQuery(database, name, query string) error
CreateDatabase(name string) (*meta.DatabaseInfo, error)
CreateDatabaseWithRetentionPolicy(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error)
CreateRetentionPolicy(database string, spec *meta.RetentionPolicySpec) (*meta.RetentionPolicyInfo, error)
CreateRetentionPolicy(database string, spec *meta.RetentionPolicySpec, makeDefault bool) (*meta.RetentionPolicyInfo, error)
CreateSubscription(database, rp, name, mode string, destinations []string) error
CreateUser(name, password string, admin bool) (*meta.UserInfo, error)
Database(name string) *meta.DatabaseInfo
@ -25,10 +25,9 @@ type MetaClient interface {
DropUser(name string) error
RetentionPolicy(database, name string) (rpi *meta.RetentionPolicyInfo, err error)
SetAdminPrivilege(username string, admin bool) error
SetDefaultRetentionPolicy(database, name string) error
SetPrivilege(username, database string, p influxql.Privilege) error
ShardsByTimeRange(sources influxql.Sources, tmin, tmax time.Time) (a []meta.ShardInfo, err error)
UpdateRetentionPolicy(database, name string, rpu *meta.RetentionPolicyUpdate) error
UpdateRetentionPolicy(database, name string, rpu *meta.RetentionPolicyUpdate, makeDefault bool) error
UpdateUser(name, password string) error
UserPrivilege(username, database string) (*influxql.Privilege, error)
UserPrivileges(username string) (map[string]influxql.Privilege, error)

17
coordinator/meta_client_test.go
View File

@ -12,7 +12,7 @@ type MetaClient struct {
CreateContinuousQueryFn func(database, name, query string) error
CreateDatabaseFn func(name string) (*meta.DatabaseInfo, error)
CreateDatabaseWithRetentionPolicyFn func(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error)
CreateRetentionPolicyFn func(database string, spec *meta.RetentionPolicySpec) (*meta.RetentionPolicyInfo, error)
CreateRetentionPolicyFn func(database string, spec *meta.RetentionPolicySpec, makeDefault bool) (*meta.RetentionPolicyInfo, error)
CreateSubscriptionFn func(database, rp, name, mode string, destinations []string) error
CreateUserFn func(name, password string, admin bool) (*meta.UserInfo, error)
DatabaseFn func(name string) *meta.DatabaseInfo
@ -30,10 +30,9 @@ type MetaClient struct {
MetaNodesFn func() ([]meta.NodeInfo, error)
RetentionPolicyFn func(database, name string) (rpi *meta.RetentionPolicyInfo, err error)
SetAdminPrivilegeFn func(username string, admin bool) error
SetDefaultRetentionPolicyFn func(database, name string) error
SetPrivilegeFn func(username, database string, p influxql.Privilege) error
ShardsByTimeRangeFn func(sources influxql.Sources, tmin, tmax time.Time) (a []meta.ShardInfo, err error)
UpdateRetentionPolicyFn func(database, name string, rpu *meta.RetentionPolicyUpdate) error
UpdateRetentionPolicyFn func(database, name string, rpu *meta.RetentionPolicyUpdate, makeDefault bool) error
UpdateUserFn func(name, password string) error
UserPrivilegeFn func(username, database string) (*influxql.Privilege, error)
UserPrivilegesFn func(username string) (map[string]influxql.Privilege, error)
@ -52,8 +51,8 @@ func (c *MetaClient) CreateDatabaseWithRetentionPolicy(name string, spec *meta.R
return c.CreateDatabaseWithRetentionPolicyFn(name, spec)
}
func (c *MetaClient) CreateRetentionPolicy(database string, spec *meta.RetentionPolicySpec) (*meta.RetentionPolicyInfo, error) {
return c.CreateRetentionPolicyFn(database, spec)
func (c *MetaClient) CreateRetentionPolicy(database string, spec *meta.RetentionPolicySpec, makeDefault bool) (*meta.RetentionPolicyInfo, error) {
return c.CreateRetentionPolicyFn(database, spec, makeDefault)
}
func (c *MetaClient) DropShard(id uint64) error {
@ -124,10 +123,6 @@ func (c *MetaClient) SetAdminPrivilege(username string, admin bool) error {
return c.SetAdminPrivilegeFn(username, admin)
}
func (c *MetaClient) SetDefaultRetentionPolicy(database, name string) error {
return c.SetDefaultRetentionPolicyFn(database, name)
}
func (c *MetaClient) SetPrivilege(username, database string, p influxql.Privilege) error {
return c.SetPrivilegeFn(username, database, p)
}
@ -136,8 +131,8 @@ func (c *MetaClient) ShardsByTimeRange(sources influxql.Sources, tmin, tmax time
return c.ShardsByTimeRangeFn(sources, tmin, tmax)
}
func (c *MetaClient) UpdateRetentionPolicy(database, name string, rpu *meta.RetentionPolicyUpdate) error {
return c.UpdateRetentionPolicyFn(database, name, rpu)
func (c *MetaClient) UpdateRetentionPolicy(database, name string, rpu *meta.RetentionPolicyUpdate, makeDefault bool) error {
return c.UpdateRetentionPolicyFn(database, name, rpu, makeDefault)
}
func (c *MetaClient) UpdateUser(name, password string) error {

18
coordinator/statement_executor.go
View File

@ -203,17 +203,9 @@ func (e *StatementExecutor) executeAlterRetentionPolicyStatement(stmt *influxql.
}
// Update the retention policy.
if err := e.MetaClient.UpdateRetentionPolicy(stmt.Database, stmt.Name, rpu); err != nil {
if err := e.MetaClient.UpdateRetentionPolicy(stmt.Database, stmt.Name, rpu, stmt.Default); err != nil {
return err
}
// If requested, set as default retention policy.
if stmt.Default {
if err := e.MetaClient.SetDefaultRetentionPolicy(stmt.Database, stmt.Name); err != nil {
return err
}
}
return nil
}
@ -271,17 +263,11 @@ func (e *StatementExecutor) executeCreateRetentionPolicyStatement(stmt *influxql
}
// Create new retention policy.
rp, err := e.MetaClient.CreateRetentionPolicy(stmt.Database, &spec)
_, err := e.MetaClient.CreateRetentionPolicy(stmt.Database, &spec, stmt.Default)
if err != nil {
return err
}
// If requested, set new policy as the default.
if stmt.Default {
if err := e.MetaClient.SetDefaultRetentionPolicy(stmt.Database, rp.Name); err != nil {
return err
}
}
return nil
}

35
internal/meta_client.go
View File

@ -13,7 +13,7 @@ type MetaClientMock struct {
CreateContinuousQueryFn func(database, name, query string) error
CreateDatabaseFn func(name string) (*meta.DatabaseInfo, error)
CreateDatabaseWithRetentionPolicyFn func(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error)
CreateRetentionPolicyFn func(database string, spec *meta.RetentionPolicySpec) (*meta.RetentionPolicyInfo, error)
CreateRetentionPolicyFn func(database string, spec *meta.RetentionPolicySpec, makeDefault bool) (*meta.RetentionPolicyInfo, error)
CreateShardGroupFn func(database, policy string, timestamp time.Time) (*meta.ShardGroupInfo, error)
CreateSubscriptionFn func(database, rp, name, mode string, destinations []string) error
CreateUserFn func(name, password string, admin bool) (*meta.UserInfo, error)
@ -34,17 +34,16 @@ type MetaClientMock struct {
RetentionPolicyFn func(database, name string) (rpi *meta.RetentionPolicyInfo, err error)
SetAdminPrivilegeFn func(username string, admin bool) error
SetDataFn func(*meta.Data) error
SetDefaultRetentionPolicyFn func(database, name string) error
SetPrivilegeFn func(username, database string, p influxql.Privilege) error
ShardsByTimeRangeFn func(sources influxql.Sources, tmin, tmax time.Time) (a []meta.ShardInfo, err error)
ShardOwnerFn func(shardID uint64) (database, policy string, sgi *meta.ShardGroupInfo)
UpdateRetentionPolicyFn func(database, name string, rpu *meta.RetentionPolicyUpdate) error
UpdateUserFn func(name, password string) error
UserPrivilegeFn func(username, database string) (*influxql.Privilege, error)
UserPrivilegesFn func(username string) (map[string]influxql.Privilege, error)
UsersFn func() []meta.UserInfo
SetAdminPrivilegeFn func(username string, admin bool) error
SetDataFn func(*meta.Data) error
SetPrivilegeFn func(username, database string, p influxql.Privilege) error
ShardsByTimeRangeFn func(sources influxql.Sources, tmin, tmax time.Time) (a []meta.ShardInfo, err error)
ShardOwnerFn func(shardID uint64) (database, policy string, sgi *meta.ShardGroupInfo)
UpdateRetentionPolicyFn func(database, name string, rpu *meta.RetentionPolicyUpdate, makeDefault bool) error
UpdateUserFn func(name, password string) error
UserPrivilegeFn func(username, database string) (*influxql.Privilege, error)
UserPrivilegesFn func(username string) (map[string]influxql.Privilege, error)
UsersFn func() []meta.UserInfo
}
func (c *MetaClientMock) Close() error {
@ -63,8 +62,8 @@ func (c *MetaClientMock) CreateDatabaseWithRetentionPolicy(name string, spec *me
return c.CreateDatabaseWithRetentionPolicyFn(name, spec)
}
func (c *MetaClientMock) CreateRetentionPolicy(database string, spec *meta.RetentionPolicySpec) (*meta.RetentionPolicyInfo, error) {
return c.CreateRetentionPolicyFn(database, spec)
func (c *MetaClientMock) CreateRetentionPolicy(database string, spec *meta.RetentionPolicySpec, makeDefault bool) (*meta.RetentionPolicyInfo, error) {
return c.CreateRetentionPolicyFn(database, spec, makeDefault)
}
func (c *MetaClientMock) CreateShardGroup(database, policy string, timestamp time.Time) (*meta.ShardGroupInfo, error) {
@ -123,10 +122,6 @@ func (c *MetaClientMock) SetAdminPrivilege(username string, admin bool) error {
return c.SetAdminPrivilegeFn(username, admin)
}
func (c *MetaClientMock) SetDefaultRetentionPolicy(database, name string) error {
return c.SetDefaultRetentionPolicyFn(database, name)
}
func (c *MetaClientMock) SetPrivilege(username, database string, p influxql.Privilege) error {
return c.SetPrivilegeFn(username, database, p)
}
@ -139,8 +134,8 @@ func (c *MetaClientMock) ShardOwner(shardID uint64) (database, policy string, sg
return c.ShardOwnerFn(shardID)
}
func (c *MetaClientMock) UpdateRetentionPolicy(database, name string, rpu *meta.RetentionPolicyUpdate) error {
return c.UpdateRetentionPolicyFn(database, name, rpu)
func (c *MetaClientMock) UpdateRetentionPolicy(database, name string, rpu *meta.RetentionPolicyUpdate, makeDefault bool) error {
return c.UpdateRetentionPolicyFn(database, name, rpu, makeDefault)
}
func (c *MetaClientMock) UpdateUser(name, password string) error {

4
services/graphite/service.go
View File

@ -85,7 +85,7 @@ type Service struct {
MetaClient interface {
CreateDatabase(name string) (*meta.DatabaseInfo, error)
CreateDatabaseWithRetentionPolicy(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error)
CreateRetentionPolicy(database string, spec *meta.RetentionPolicySpec) (*meta.RetentionPolicyInfo, error)
CreateRetentionPolicy(database string, spec *meta.RetentionPolicySpec, makeDefault bool) (*meta.RetentionPolicyInfo, error)
Database(name string) *meta.DatabaseInfo
RetentionPolicy(database, name string) (*meta.RetentionPolicyInfo, error)
}
@ -234,7 +234,7 @@ func (s *Service) createInternalStorage() error {
if db := s.MetaClient.Database(s.database); db != nil {
if rp, _ := s.MetaClient.RetentionPolicy(s.database, s.retentionPolicy); rp == nil {
spec := meta.RetentionPolicySpec{Name: s.retentionPolicy}
if _, err := s.MetaClient.CreateRetentionPolicy(s.database, &spec); err != nil {
if _, err := s.MetaClient.CreateRetentionPolicy(s.database, &spec, true); err != nil {
return err
}
}

2
services/graphite/service_test.go
View File

@ -271,7 +271,7 @@ func NewTestService(c *Config) *TestService {
MetaClient: &internal.MetaClientMock{},
}
service.MetaClient.CreateRetentionPolicyFn = func(string, *meta.RetentionPolicySpec) (*meta.RetentionPolicyInfo, error) {
service.MetaClient.CreateRetentionPolicyFn = func(string, *meta.RetentionPolicySpec, bool) (*meta.RetentionPolicyInfo, error) {
return nil, nil
}

43
services/meta/client.go
View File

@ -186,10 +186,7 @@ func (c *Client) CreateDatabase(name string) (*DatabaseInfo, error) {
// create default retention policy
if c.retentionAutoCreate {
rpi := DefaultRetentionPolicyInfo()
if err := data.CreateRetentionPolicy(name, rpi); err != nil {
return nil, err
}
if err := data.SetDefaultRetentionPolicy(name, rpi.Name); err != nil {
if err := data.CreateRetentionPolicy(name, rpi, true); err != nil {
return nil, err
}
}
@ -224,7 +221,7 @@ func (c *Client) CreateDatabaseWithRetentionPolicy(name string, spec *RetentionP
rpi := spec.NewRetentionPolicyInfo()
if rp := db.RetentionPolicy(rpi.Name); rp == nil {
if err := data.CreateRetentionPolicy(name, rpi); err != nil {
if err := data.CreateRetentionPolicy(name, rpi, true); err != nil {
return nil, err
}
} else if !spec.Matches(rp) {
@ -237,20 +234,18 @@ func (c *Client) CreateDatabaseWithRetentionPolicy(name string, spec *RetentionP
// policy we just created. If the default is different from what we are
// trying to create, record it as a conflict and abandon with an error.
if db.DefaultRetentionPolicy == "" {
if err := data.SetDefaultRetentionPolicy(name, rpi.Name); err != nil {
return nil, err
}
db.DefaultRetentionPolicy = rpi.Name
} else if rpi.Name != db.DefaultRetentionPolicy {
return nil, ErrRetentionPolicyConflict
}
// Refresh the database info.
db = data.Database(name)
if err := c.commit(data); err != nil {
return nil, err
}
// Refresh the database info.
db = data.Database(name)
return db, nil
}
@ -273,7 +268,7 @@ func (c *Client) DropDatabase(name string) error {
}
// CreateRetentionPolicy creates a retention policy on the specified database.
func (c *Client) CreateRetentionPolicy(database string, spec *RetentionPolicySpec) (*RetentionPolicyInfo, error) {
func (c *Client) CreateRetentionPolicy(database string, spec *RetentionPolicySpec, makeDefault bool) (*RetentionPolicyInfo, error) {
c.mu.Lock()
defer c.mu.Unlock()
@ -284,7 +279,7 @@ func (c *Client) CreateRetentionPolicy(database string, spec *RetentionPolicySpe
}
rp := spec.NewRetentionPolicyInfo()
if err := data.CreateRetentionPolicy(database, rp); err != nil {
if err := data.CreateRetentionPolicy(database, rp, makeDefault); err != nil {
return nil, err
}
@ -326,32 +321,14 @@ func (c *Client) DropRetentionPolicy(database, name string) error {
return nil
}
// SetDefaultRetentionPolicy sets a database's default retention policy.
func (c *Client) SetDefaultRetentionPolicy(database, name string) error {
c.mu.Lock()
defer c.mu.Unlock()
data := c.cacheData.Clone()
if err := data.SetDefaultRetentionPolicy(database, name); err != nil {
return err
}
if err := c.commit(data); err != nil {
return err
}
return nil
}
// UpdateRetentionPolicy updates a retention policy.
func (c *Client) UpdateRetentionPolicy(database, name string, rpu *RetentionPolicyUpdate) error {
func (c *Client) UpdateRetentionPolicy(database, name string, rpu *RetentionPolicyUpdate, makeDefault bool) error {
c.mu.Lock()
defer c.mu.Unlock()
data := c.cacheData.Clone()
if err := data.UpdateRetentionPolicy(database, name, rpu); err != nil {
if err := data.UpdateRetentionPolicy(database, name, rpu, makeDefault); err != nil {
return err
}

26
services/meta/client_test.go
View File

@ -248,7 +248,7 @@ func TestMetaClient_CreateRetentionPolicy(t *testing.T) {
ReplicaN: &rp0.ReplicaN,
Duration: &rp0.Duration,
ShardGroupDuration: rp0.ShardGroupDuration,
}); err != nil {
}, true); err != nil {
t.Fatal(err)
}
@ -265,7 +265,7 @@ func TestMetaClient_CreateRetentionPolicy(t *testing.T) {
ReplicaN: &rp0.ReplicaN,
Duration: &rp0.Duration,
ShardGroupDuration: rp0.ShardGroupDuration,
}); err != nil {
}, true); err != nil {
t.Fatal(err)
} else if actual, err = c.RetentionPolicy("db0", "rp0"); err != nil {
t.Fatal(err)
@ -283,7 +283,7 @@ func TestMetaClient_CreateRetentionPolicy(t *testing.T) {
ReplicaN: &rp1.ReplicaN,
Duration: &rp1.Duration,
ShardGroupDuration: rp1.ShardGroupDuration,
})
}, true)
if exp := meta.ErrRetentionPolicyExists; got != exp {
t.Fatalf("got error %v, expected error %v", got, exp)
}
@ -298,7 +298,7 @@ func TestMetaClient_CreateRetentionPolicy(t *testing.T) {
ReplicaN: &rp1.ReplicaN,
Duration: &rp1.Duration,
ShardGroupDuration: rp1.ShardGroupDuration,
})
}, true)
if exp := meta.ErrRetentionPolicyExists; got != exp {
t.Fatalf("got error %v, expected error %v", got, exp)
}
@ -313,7 +313,7 @@ func TestMetaClient_CreateRetentionPolicy(t *testing.T) {
ReplicaN: &rp1.ReplicaN,
Duration: &rp1.Duration,
ShardGroupDuration: rp1.ShardGroupDuration,
})
}, true)
if exp := meta.ErrRetentionPolicyExists; got != exp {
t.Fatalf("got error %v, expected error %v", got, exp)
}
@ -329,13 +329,13 @@ func TestMetaClient_CreateRetentionPolicy(t *testing.T) {
ReplicaN: &rp1.ReplicaN,
Duration: &rp1.Duration,
ShardGroupDuration: rp1.ShardGroupDuration,
})
}, true)
if exp := meta.ErrIncompatibleDurations; got != exp {
t.Fatalf("got error %v, expected error %v", got, exp)
}
}
func TestMetaClient_SetDefaultRetentionPolicy(t *testing.T) {
func TestMetaClient_DefaultRetentionPolicy(t *testing.T) {
t.Parallel()
d, c := newClient()
@ -402,7 +402,7 @@ func TestMetaClient_UpdateRetentionPolicy(t *testing.T) {
if err := c.UpdateRetentionPolicy("db0", "rp0", &meta.RetentionPolicyUpdate{
Duration: &duration,
ReplicaN: &replicaN,
}); err != nil {
}, true); err != nil {
t.Fatal(err)
}
@ -418,7 +418,7 @@ func TestMetaClient_UpdateRetentionPolicy(t *testing.T) {
duration = rpi.ShardGroupDuration / 2
if err := c.UpdateRetentionPolicy("db0", "rp0", &meta.RetentionPolicyUpdate{
Duration: &duration,
}); err == nil {
}, true); err == nil {
t.Fatal("expected error")
} else if err != meta.ErrIncompatibleDurations {
t.Fatalf("expected error '%s', got '%s'", meta.ErrIncompatibleDurations, err)
@ -428,7 +428,7 @@ func TestMetaClient_UpdateRetentionPolicy(t *testing.T) {
sgDuration := rpi.Duration * 2
if err := c.UpdateRetentionPolicy("db0", "rp0", &meta.RetentionPolicyUpdate{
ShardGroupDuration: &sgDuration,
}); err == nil {
}, true); err == nil {
t.Fatal("expected error")
} else if err != meta.ErrIncompatibleDurations {
t.Fatalf("expected error '%s', got '%s'", meta.ErrIncompatibleDurations, err)
@ -440,7 +440,7 @@ func TestMetaClient_UpdateRetentionPolicy(t *testing.T) {
if err := c.UpdateRetentionPolicy("db0", "rp0", &meta.RetentionPolicyUpdate{
Duration: &duration,
ShardGroupDuration: &sgDuration,
}); err == nil {
}, true); err == nil {
t.Fatal("expected error")
} else if err != meta.ErrIncompatibleDurations {
t.Fatalf("expected error '%s', got '%s'", meta.ErrIncompatibleDurations, err)
@ -452,7 +452,7 @@ func TestMetaClient_UpdateRetentionPolicy(t *testing.T) {
if err := c.UpdateRetentionPolicy("db0", "rp0", &meta.RetentionPolicyUpdate{
Duration: &duration,
ShardGroupDuration: &sgDuration,
}); err != nil {
}, true); err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
@ -481,7 +481,7 @@ func TestMetaClient_DropRetentionPolicy(t *testing.T) {
Name: "rp0",
Duration: &duration,
ReplicaN: &replicaN,
}); err != nil {
}, true); err != nil {
t.Fatal(err)
}

34
services/meta/data.go
View File

@ -135,7 +135,7 @@ func (data *Data) RetentionPolicy(database, name string) (*RetentionPolicyInfo,
// CreateRetentionPolicy creates a new retention policy on a database.
// Returns an error if name is blank or if a database does not exist.
func (data *Data) CreateRetentionPolicy(database string, rpi *RetentionPolicyInfo) error {
func (data *Data) CreateRetentionPolicy(database string, rpi *RetentionPolicyInfo, makeDefault bool) error {
// Validate retention policy.
if rpi == nil {
return ErrRetentionPolicyRequired
@ -163,11 +163,21 @@ func (data *Data) CreateRetentionPolicy(database string, rpi *RetentionPolicyInf
if rp.ReplicaN != rpi.ReplicaN || rp.Duration != rpi.Duration || rp.ShardGroupDuration != rpi.ShardGroupDuration {
return ErrRetentionPolicyExists
}
// if they want to make it default, and it's not the default, it's not an identical command so it's an error
if makeDefault && di.DefaultRetentionPolicy != rpi.Name {
return ErrRetentionPolicyConflict
}
return nil
}
// Append copy of new policy.
di.RetentionPolicies = append(di.RetentionPolicies, *rpi)
// Set the default if needed
if makeDefault {
di.DefaultRetentionPolicy = rpi.Name
}
return nil
}
@ -212,7 +222,7 @@ func (rpu *RetentionPolicyUpdate) SetReplicaN(v int) { rpu.ReplicaN = &v }
func (rpu *RetentionPolicyUpdate) SetShardGroupDuration(v time.Duration) { rpu.ShardGroupDuration = &v }
// UpdateRetentionPolicy updates an existing retention policy.
func (data *Data) UpdateRetentionPolicy(database, name string, rpu *RetentionPolicyUpdate) error {
func (data *Data) UpdateRetentionPolicy(database, name string, rpu *RetentionPolicyUpdate, makeDefault bool) error {
// Find database.
di := data.Database(database)
if di == nil {
@ -258,26 +268,10 @@ func (data *Data) UpdateRetentionPolicy(database, name string, rpu *RetentionPol
rpi.ShardGroupDuration = normalisedShardDuration(*rpu.ShardGroupDuration, rpi.Duration)
}
return nil
}
// SetDefaultRetentionPolicy sets the default retention policy for a database.
func (data *Data) SetDefaultRetentionPolicy(database, name string) error {
if name == "" {
name = DefaultRetentionPolicyName
if di.DefaultRetentionPolicy != rpi.Name && makeDefault {
di.DefaultRetentionPolicy = rpi.Name
}
// Find database and verify policy exists.
di := data.Database(database)
if di == nil {
return influxdb.ErrDatabaseNotFound(database)
} else if di.RetentionPolicy(name) == nil {
return influxdb.ErrRetentionPolicyNotFound(name)
}
// Set default policy.
di.DefaultRetentionPolicy = name
return nil
}

56
services/meta/data_internal_test.go Normal file
View File

@ -0,0 +1,56 @@
package meta
import (
"reflect"
"sort"
"time"
"testing"
)
func Test_newShardOwner(t *testing.T) {
// An error is returned if there are no data nodes available.
_, err := NewShardOwner(ShardInfo{}, map[int]int{})
if err == nil {
t.Error("got no error, but expected one")
}
ownerFreqs := map[int]int{1: 15, 2: 11, 3: 12}
id, err := NewShardOwner(ShardInfo{ID: 4}, ownerFreqs)
if err != nil {
t.Fatal(err)
}
// The ID that owns the fewest shards is returned.
if got, exp := id, uint64(2); got != exp {
t.Errorf("got id %d, expected id %d", got, exp)
}
// The ownership frequencies are updated.
if got, exp := ownerFreqs, map[int]int{1: 15, 2: 12, 3: 12}; !reflect.DeepEqual(got, exp) {
t.Errorf("got owner frequencies %v, expected %v", got, exp)
}
}
func TestShardGroupSort(t *testing.T) {
sg1 := ShardGroupInfo{
ID: 1,
StartTime: time.Unix(1000, 0),
EndTime: time.Unix(1100, 0),
TruncatedAt: time.Unix(1050, 0),
}
sg2 := ShardGroupInfo{
ID: 2,
StartTime: time.Unix(1000, 0),
EndTime: time.Unix(1100, 0),
}
sgs := ShardGroupInfos{sg2, sg1}
sort.Sort(sgs)
if sgs[len(sgs)-1].ID != 2 {
t.Fatal("unstable sort for ShardGroupInfos")
}
}

81
services/meta/data_test.go
View File

@ -1,56 +1,55 @@
package meta
package meta_test
import (
"reflect"
"sort"
"testing"
"time"
"testing"
"github.com/influxdata/influxdb/services/meta"
)
func Test_newShardOwner(t *testing.T) {
// An error is returned if there are no data nodes available.
_, err := NewShardOwner(ShardInfo{}, map[int]int{})
if err == nil {
t.Error("got no error, but expected one")
}
func Test_Data_CreateRetentionPolicy(t *testing.T) {
data := meta.Data{}
ownerFreqs := map[int]int{1: 15, 2: 11, 3: 12}
id, err := NewShardOwner(ShardInfo{ID: 4}, ownerFreqs)
err := data.CreateDatabase("foo")
if err != nil {
t.Fatal(err)
}
// The ID that owns the fewest shards is returned.
if got, exp := id, uint64(2); got != exp {
t.Errorf("got id %d, expected id %d", got, exp)
err = data.CreateRetentionPolicy("foo", &meta.RetentionPolicyInfo{
Name: "bar",
ReplicaN: 1,
Duration: 24 * time.Hour,
}, false)
if err != nil {
t.Fatal(err)
}
// The ownership frequencies are updated.
if got, exp := ownerFreqs, map[int]int{1: 15, 2: 12, 3: 12}; !reflect.DeepEqual(got, exp) {
t.Errorf("got owner frequencies %v, expected %v", got, exp)
}
}
func TestShardGroupSort(t *testing.T) {
sg1 := ShardGroupInfo{
ID: 1,
StartTime: time.Unix(1000, 0),
EndTime: time.Unix(1100, 0),
TruncatedAt: time.Unix(1050, 0),
}
sg2 := ShardGroupInfo{
ID: 2,
StartTime: time.Unix(1000, 0),
EndTime: time.Unix(1100, 0),
}
sgs := ShardGroupInfos{sg2, sg1}
sort.Sort(sgs)
if sgs[len(sgs)-1].ID != 2 {
t.Fatal("unstable sort for ShardGroupInfos")
rp, err := data.RetentionPolicy("foo", "bar")
if err != nil {
t.Fatal(err)
}
if rp == nil {
t.Fatal("creation of retention policy failed")
}
// Try to recreate the same RP with default set to true, should fail
err = data.CreateRetentionPolicy("foo", &meta.RetentionPolicyInfo{
Name: "bar",
ReplicaN: 1,
Duration: 24 * time.Hour,
}, true)
if err == nil || err != meta.ErrRetentionPolicyConflict {
t.Fatalf("unexpected error. got: %v, exp: %s", err, meta.ErrRetentionPolicyConflict)
}
// Creating the same RP with the same specifications should succeed
err = data.CreateRetentionPolicy("foo", &meta.RetentionPolicyInfo{
Name: "bar",
ReplicaN: 1,
Duration: 24 * time.Hour,
}, false)
if err != nil {
t.Fatal(err)
}
}