<commit_before>package main

import (
	"errors"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/pivotal-cf/cf-redis-broker/brokerconfig"
	"github.com/pivotal-cf/cf-redis-broker/redis"
	"github.com/pivotal-cf/cf-redis-broker/redis/client"
	"github.com/pivotal-cf/cf-redis-broker/redisconf"
	"github.com/pivotal-cf/cf-redis-broker/s3bucket"
	"github.com/pivotal-golang/lager"

	cf_lager "github.com/cloudfoundry-incubator/cf-lager"
)

var logger = cf_lager.New("backup")

func main() {
	config, err := brokerconfig.ParseConfig(configPath())
	if err != nil {
		log.Fatal(err)
	}

	if !config.RedisConfiguration.BackupConfiguration.Enabled() {
		return
	}

	s3Client := s3bucket.NewClient(
		config.RedisConfiguration.BackupConfiguration.EndpointUrl,
		config.RedisConfiguration.BackupConfiguration.S3Region,
		config.RedisConfiguration.BackupConfiguration.AccessKeyId,
		config.RedisConfiguration.BackupConfiguration.SecretAccessKey,
	)

	bucket, err := s3Client.GetOrCreate(config.RedisConfiguration.BackupConfiguration.BucketName)
	if err != nil {
		log.Fatal(err)
	}

	instanceDirs, err := ioutil.ReadDir(config.RedisConfiguration.InstanceDataDirectory)
	if err != nil {
		log.Fatal(err)
	}

	backupErrors := []error{}
	for _, instanceDir := range instanceDirs {
		if strings.HasPrefix(instanceDir.Name(), ".") {
			continue
		}
		err = backupInstance(instanceDir, config, bucket)
		if err != nil {
			backupErrors = append(backupErrors, err)
			logger.Error("error backing up instance", err, lager.Data{
				"instance_id": instanceDir.Name(),
			})
		}
	}

	if len(backupErrors) > 0 {
		os.Exit(1)
	}
}

func backupInstance(instanceDir os.FileInfo, config brokerconfig.Config, bucket s3bucket.Bucket) error {
	pathToInstanceDirectory := filepath.Join(config.RedisConfiguration.InstanceDataDirectory, instanceDir.Name())
	if !fileExists(pathToInstanceDirectory) {
		logger.Info("instance directory not found, skipping instance backup", lager.Data{
			"Local file": pathToInstanceDirectory,
		})
		return nil
	}

	err := saveAndWaitUntilFinished(instanceDir, config)
	if err != nil {
		return err
	}

	pathToRdbFile := filepath.Join(config.RedisConfiguration.InstanceDataDirectory, instanceDir.Name(), "db", "dump.rdb")
	if !fileExists(pathToRdbFile) {
		logger.Info("dump.rb not found, skipping instance backup", lager.Data{
			"Local file": pathToRdbFile,
		})
		return nil
	}

	rdbBytes, err := ioutil.ReadFile(pathToRdbFile)
	if err != nil {
		return err
	}

	remotePath := fmt.Sprintf("%s/%s", config.RedisConfiguration.BackupConfiguration.Path, instanceDir.Name())

	logger.Info("Backing up instance", lager.Data{
		"Local file":  pathToRdbFile,
		"Remote file": remotePath,
	})

	return bucket.Upload(rdbBytes, remotePath)
}

func fileExists(path string) bool {
	if _, err := os.Stat(path); os.IsNotExist(err) {
		return false
	}
	return true
}

func saveAndWaitUntilFinished(instanceDir os.FileInfo, config brokerconfig.Config) error {

	instanceID := instanceDir.Name()

	client, err := buildRedisClient(instanceID, config)
	if err != nil {
		return err
	}

	lastSaveTime, err := client.LastRDBSaveTime()
	if err != nil {
		return err
	}

	// ensure new save time can't match last save time
	time.Sleep(time.Second)

	err = client.RunBGSave()
	if err != nil {
		return err
	}

	// wait for save to complete
	timeout := config.RedisConfiguration.BackupConfiguration.BGSaveTimeoutSeconds
	for i := 0; i < timeout; i++ {
		saveTime, err := client.LastRDBSaveTime()
		if err != nil {
			return err
		}

		if saveTime > lastSaveTime {
			return nil
		}

		time.Sleep(time.Second)
	}

	return errors.New("Timed out waiting for background save to complete")
}

func buildRedisClient(instanceID string, config brokerconfig.Config) (*client.Client, error) {

	localRepo := redis.LocalRepository{RedisConf: config.RedisConfiguration}
	instance, err := localRepo.FindByID(instanceID)
	if err != nil {
		return nil, err
	}

	instanceConf, err := redisconf.Load(localRepo.InstanceConfigPath(instanceID))
	if err != nil {
		return nil, err
	}

	return client.Connect(instance.Host, uint(instance.Port), instance.Password, instanceConf)
}

func configPath() string {
	brokerConfigYamlPath := os.Getenv("BROKER_CONFIG_PATH")
	if brokerConfigYamlPath == "" {
		panic("BROKER_CONFIG_PATH not set")
	}<commit_msg>Encapsulate creating the snapshot into the redis client<commit_after>package main

import (
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"path/filepath"
	"strings"

	"github.com/pivotal-cf/cf-redis-broker/brokerconfig"
	"github.com/pivotal-cf/cf-redis-broker/redis"
	"github.com/pivotal-cf/cf-redis-broker/redis/client"
	"github.com/pivotal-cf/cf-redis-broker/redisconf"
	"github.com/pivotal-cf/cf-redis-broker/s3bucket"
	"github.com/pivotal-golang/lager"

	cf_lager "github.com/cloudfoundry-incubator/cf-lager"
)

var logger = cf_lager.New("backup")

func main() {
	config, err := brokerconfig.ParseConfig(configPath())
	if err != nil {
		log.Fatal(err)
	}

	if !config.RedisConfiguration.BackupConfiguration.Enabled() {
		return
	}

	s3Client := s3bucket.NewClient(
		config.RedisConfiguration.BackupConfiguration.EndpointUrl,
		config.RedisConfiguration.BackupConfiguration.S3Region,
		config.RedisConfiguration.BackupConfiguration.AccessKeyId,
		config.RedisConfiguration.BackupConfiguration.SecretAccessKey,
	)

	bucket, err := s3Client.GetOrCreate(config.RedisConfiguration.BackupConfiguration.BucketName)
	if err != nil {
		log.Fatal(err)
	}

	instanceDirs, err := ioutil.ReadDir(config.RedisConfiguration.InstanceDataDirectory)
	if err != nil {
		log.Fatal(err)
	}

	backupErrors := []error{}
	for _, instanceDir := range instanceDirs {
		if strings.HasPrefix(instanceDir.Name(), ".") {
			continue
		}
		err = backupInstance(instanceDir, config, bucket)
		if err != nil {
			backupErrors = append(backupErrors, err)
			logger.Error("error backing up instance", err, lager.Data{
				"instance_id": instanceDir.Name(),
			})
		}
	}

	if len(backupErrors) > 0 {
		os.Exit(1)
	}
}

func backupInstance(instanceDir os.FileInfo, config brokerconfig.Config, bucket s3bucket.Bucket) error {
	pathToInstanceDirectory := filepath.Join(config.RedisConfiguration.InstanceDataDirectory, instanceDir.Name())
	if !fileExists(pathToInstanceDirectory) {
		logger.Info("instance directory not found, skipping instance backup", lager.Data{
			"Local file": pathToInstanceDirectory,
		})
		return nil
	}

	err := saveAndWaitUntilFinished(instanceDir, config)
	if err != nil {
		return err
	}

	pathToRdbFile := filepath.Join(config.RedisConfiguration.InstanceDataDirectory, instanceDir.Name(), "db", "dump.rdb")
	if !fileExists(pathToRdbFile) {
		logger.Info("dump.rb not found, skipping instance backup", lager.Data{
			"Local file": pathToRdbFile,
		})
		return nil
	}

	rdbBytes, err := ioutil.ReadFile(pathToRdbFile)
	if err != nil {
		return err
	}

	remotePath := fmt.Sprintf("%s/%s", config.RedisConfiguration.BackupConfiguration.Path, instanceDir.Name())

	logger.Info("Backing up instance", lager.Data{
		"Local file":  pathToRdbFile,
		"Remote file": remotePath,
	})

	return bucket.Upload(rdbBytes, remotePath)
}

func fileExists(path string) bool {
	if _, err := os.Stat(path); os.IsNotExist(err) {
		return false
	}
	return true
}

func saveAndWaitUntilFinished(instanceDir os.FileInfo, config brokerconfig.Config) error {

	instanceID := instanceDir.Name()

	client, err := buildRedisClient(instanceID, config)
	if err != nil {
		return err
	}

	return client.CreateSnapshot(config.RedisConfiguration.BackupConfiguration.BGSaveTimeoutSeconds)
}

func buildRedisClient(instanceID string, config brokerconfig.Config) (*client.Client, error) {

	localRepo := redis.LocalRepository{RedisConf: config.RedisConfiguration}
	instance, err := localRepo.FindByID(instanceID)
	if err != nil {
		return nil, err
	}

	instanceConf, err := redisconf.Load(localRepo.InstanceConfigPath(instanceID))
	if err != nil {
		return nil, err
	}

	return client.Connect(instance.Host, uint(instance.Port), instance.Password, instanceConf)
}

func configPath() string {
	brokerConfigYamlPath := os.Getenv("BROKER_CONFIG_PATH")
	if brokerConfigYamlPath == "" {
		panic("BROKER_CONFIG_PATH not set")
	}
{'fraction_non_alphanumeric': 0.12373521882238206, 'fraction_numerical': 0.0020724125320004875, 'mean_word_length': 4.868383404864091, 'pattern_counts': {'":': 10, '<': 4, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1411395', 'n_tokens_mistral': 2929, 'n_tokens_neox': 2782, 'n_words': 538}