Merge pull request #1761 from kmlebedev/todaysDateFormat

replication to todays date directory
This commit is contained in:
Chris Lu 2021-01-27 00:15:16 -08:00 committed by GitHub
commit 2266b5c528
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 20 additions and 8 deletions

View file

@ -11,6 +11,7 @@ enabled = true
# This URL will Dial the RabbitMQ server at the URL in the environment
# variable RABBIT_SERVER_URL and open the exchange "myexchange".
# The exchange must have already been created by some other means, like
# the RabbitMQ management plugin.
# the RabbitMQ management plugin. Сreate myexchange of type fanout and myqueue then
# create binding myexchange => myqueue
topic_url = "rabbit://swexchange"
sub_url = "rabbit://swqueue"

View file

@ -9,3 +9,4 @@ directory = "/buckets"
[sink.local]
enabled = true
directory = "/data"
todays_date_format = "2006-02-01"

View file

@ -329,7 +329,8 @@ enabled = false
# This URL will Dial the RabbitMQ server at the URL in the environment
# variable RABBIT_SERVER_URL and open the exchange "myexchange".
# The exchange must have already been created by some other means, like
# the RabbitMQ management plugin.
# the RabbitMQ management plugin. Сreate myexchange of type fanout and myqueue then
# create binding myexchange => myqueue
topic_url = "rabbit://myexchange"
sub_url = "rabbit://myqueue"
`
@ -353,6 +354,8 @@ directory = "/buckets"
[sink.local]
enabled = false
directory = "/backup"
# all replicated files are under todays date directory tree
todays_date_format = ""
[sink.filer]
enabled = false

View file

@ -12,10 +12,12 @@ import (
"os"
"path/filepath"
"strings"
"time"
)
type LocalSink struct {
dir string
todaysDateFormat string
filerSource *source.FilerSource
}
@ -35,18 +37,23 @@ func (localsink *LocalSink) isMultiPartEntry(key string) bool {
return strings.HasSuffix(key, ".part") && strings.Contains(key, "/.uploads/")
}
func (localsink *LocalSink) initialize(dir string) error {
func (localsink *LocalSink) initialize(dir string, todaysDateFormat string) error {
localsink.dir = dir
localsink.todaysDateFormat = todaysDateFormat
return nil
}
func (localsink *LocalSink) Initialize(configuration util.Configuration, prefix string) error {
dir := configuration.GetString(prefix + "directory")
todaysDateFormat := configuration.GetString(prefix + "todays_date_format")
glog.V(4).Infof("sink.local.directory: %v", dir)
return localsink.initialize(dir)
return localsink.initialize(dir, todaysDateFormat)
}
func (localsink *LocalSink) GetSinkToDirectory() string {
if localsink.todaysDateFormat != "" {
return filepath.Join(localsink.dir, time.Now().Format(localsink.todaysDateFormat))
}
return localsink.dir
}
@ -56,7 +63,7 @@ func (localsink *LocalSink) DeleteEntry(key string, isDirectory, deleteIncludeCh
}
glog.V(4).Infof("Delete Entry key: %s", key)
if err := os.Remove(key); err != nil {
return err
glog.V(0).Infof("remove entry key %s: %s", key, err)
}
return nil
}