Support specifying a retention policy for the graphite service

The graphite service will attempt to create the retention policy and use
it. If the retention policy doesn't exist, it will be created with the
default options.

Fixes #5655.

CHANGELOG.md

@@ -22,6 +22,7 @@
 - [#6686](https://github.com/influxdata/influxdb/pull/6686): Optimize timestamp run-length decoding
 - [#6713](https://github.com/influxdata/influxdb/pull/6713): Reduce allocations during query parsing.
 - [#3733](https://github.com/influxdata/influxdb/issues/3733): Modify the default retention policy name and make it configurable.
+- [#5655](https://github.com/influxdata/influxdb/issues/5655): Support specifying a retention policy for the graphite service.
 
 ### Bugfixes
 

services/graphite/config.go

@@ -54,6 +54,7 @@ type Config struct {
 	Enabled          bool          `toml:"enabled"`
 	BindAddress      string        `toml:"bind-address"`
 	Database         string        `toml:"database"`
+	RetentionPolicy  string        `toml:"retention-policy"`
 	Protocol         string        `toml:"protocol"`
 	BatchSize        int           `toml:"batch-size"`
 	BatchPending     int           `toml:"batch-pending"`

services/graphite/config_test.go

@@ -14,6 +14,7 @@ func TestConfig_Parse(t *testing.T) {
 	if _, err := toml.Decode(`
 bind-address = ":8080"
 database = "mydb"
+retention-policy = "myrp"
 enabled = true
 protocol = "tcp"
 batch-size=100
@@ -31,6 +32,8 @@ tags=["region=us-east"]
 		t.Fatalf("unexpected bind address: %s", c.BindAddress)
 	} else if c.Database != "mydb" {
 		t.Fatalf("unexpected database selected: %s", c.Database)
+	} else if c.RetentionPolicy != "myrp" {
+		t.Fatalf("unexpected retention policy selected: %s", c.RetentionPolicy)
 	} else if c.Enabled != true {
 		t.Fatalf("unexpected graphite enabled: %v", c.Enabled)
 	} else if c.Protocol != "tcp" {

services/graphite/service.go

@@ -54,6 +54,7 @@ type Service struct {
 
 	bindAddress     string
 	database        string
+	retentionPolicy string
 	protocol        string
 	batchSize       int
 	batchPending    int
@@ -85,6 +86,10 @@ type Service struct {
 	}
 	MetaClient interface {
 		CreateDatabase(name string) (*meta.DatabaseInfo, error)
+		CreateDatabaseWithRetentionPolicy(name string, rpi *meta.RetentionPolicyInfo) (*meta.DatabaseInfo, error)
+		CreateRetentionPolicy(database string, rpi *meta.RetentionPolicyInfo) (*meta.RetentionPolicyInfo, error)
+		Database(name string) *meta.DatabaseInfo
+		RetentionPolicy(database, name string) (*meta.RetentionPolicyInfo, error)
 	}
 }
 
@@ -96,6 +101,7 @@ func NewService(c Config) (*Service, error) {
 	s := Service{
 		bindAddress:     d.BindAddress,
 		database:        d.Database,
+		retentionPolicy: d.RetentionPolicy,
 		protocol:        d.Protocol,
 		batchSize:       d.BatchSize,
 		batchPending:    d.BatchPending,
@@ -137,10 +143,20 @@ func (s *Service) Open() error {
 		s.Monitor.RegisterDiagnosticsClient(s.diagsKey, s)
 	}
 
-	if _, err := s.MetaClient.CreateDatabase(s.database); err != nil {
+	if db := s.MetaClient.Database(s.database); db != nil {
+		if rp, _ := s.MetaClient.RetentionPolicy(s.database, s.retentionPolicy); rp == nil {
+			rpi := meta.NewRetentionPolicyInfo(s.retentionPolicy)
+			if _, err := s.MetaClient.CreateRetentionPolicy(s.database, rpi); err != nil {
+				s.logger.Printf("Failed to ensure target retention policy %s exists: %s", s.database, err.Error())
+			}
+		}
+	} else {
+		rpi := meta.NewRetentionPolicyInfo(s.retentionPolicy)
+		if _, err := s.MetaClient.CreateDatabaseWithRetentionPolicy(s.database, rpi); err != nil {
 			s.logger.Printf("Failed to ensure target database %s exists: %s", s.database, err.Error())
 			return err
 		}
+	}
 
 	s.batcher = tsdb.NewPointBatcher(s.batchSize, s.batchPending, s.batchTimeout)
 	s.batcher.Start()
@@ -355,7 +371,7 @@ func (s *Service) processBatches(batcher *tsdb.PointBatcher) {
 	for {
 		select {
 		case batch := <-batcher.Out():
-			if err := s.PointsWriter.WritePoints(s.database, "", models.ConsistencyLevelAny, batch); err == nil {
+			if err := s.PointsWriter.WritePoints(s.database, s.retentionPolicy, models.ConsistencyLevelAny, batch); err == nil {
 				s.statMap.Add(statBatchesTransmitted, 1)
 				s.statMap.Add(statPointsTransmitted, int64(len(batch)))
 			} else {

services/graphite/service_test.go

@@ -176,6 +176,23 @@ func (d *DatabaseCreator) CreateDatabase(name string) (*meta.DatabaseInfo, error
 	return nil, nil
 }
 
+func (d *DatabaseCreator) CreateRetentionPolicy(database string, rpi *meta.RetentionPolicyInfo) (*meta.RetentionPolicyInfo, error) {
+	return nil, nil
+}
+
+func (d *DatabaseCreator) CreateDatabaseWithRetentionPolicy(name string, rpi *meta.RetentionPolicyInfo) (*meta.DatabaseInfo, error) {
+	d.Created = true
+	return nil, nil
+}
+
+func (d *DatabaseCreator) Database(name string) *meta.DatabaseInfo {
+	return nil
+}
+
+func (d *DatabaseCreator) RetentionPolicy(database, name string) (*meta.RetentionPolicyInfo, error) {
+	return nil, nil
+}
+
 // Test Helpers
 func errstr(err error) string {
 	if err != nil {