233 lines
6.6 KiB
Go
233 lines
6.6 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"flag"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"os"
|
|
"os/exec"
|
|
|
|
"github.com/benbjohnson/litestream"
|
|
"github.com/benbjohnson/litestream/abs"
|
|
"github.com/benbjohnson/litestream/gs"
|
|
"github.com/benbjohnson/litestream/http"
|
|
"github.com/benbjohnson/litestream/s3"
|
|
"github.com/benbjohnson/litestream/sftp"
|
|
"github.com/mattn/go-shellwords"
|
|
)
|
|
|
|
// ReplicateCommand represents a command that continuously replicates SQLite databases.
|
|
type ReplicateCommand struct {
|
|
stdin io.Reader
|
|
stdout io.Writer
|
|
stderr io.Writer
|
|
|
|
configPath string
|
|
noExpandEnv bool
|
|
|
|
cmd *exec.Cmd // subcommand
|
|
execCh chan error // subcommand error channel
|
|
|
|
Config Config
|
|
|
|
server *litestream.Server
|
|
httpServer *http.Server
|
|
}
|
|
|
|
// NewReplicateCommand returns a new instance of ReplicateCommand.
|
|
func NewReplicateCommand(stdin io.Reader, stdout, stderr io.Writer) *ReplicateCommand {
|
|
return &ReplicateCommand{
|
|
stdin: stdin,
|
|
stdout: stdout,
|
|
stderr: stderr,
|
|
|
|
execCh: make(chan error),
|
|
}
|
|
}
|
|
|
|
// ParseFlags parses the CLI flags and loads the configuration file.
|
|
func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err error) {
|
|
fs := flag.NewFlagSet("litestream-replicate", flag.ContinueOnError)
|
|
execFlag := fs.String("exec", "", "execute subcommand")
|
|
addr := fs.String("addr", "", "HTTP bind address (host:port)")
|
|
registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
|
|
fs.Usage = c.Usage
|
|
if err := fs.Parse(args); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Load configuration or use CLI args to build db/replica.
|
|
if fs.NArg() == 1 {
|
|
return fmt.Errorf("must specify at least one replica URL for %s", fs.Arg(0))
|
|
} else if fs.NArg() > 1 {
|
|
if c.configPath != "" {
|
|
return fmt.Errorf("cannot specify a replica URL and the -config flag")
|
|
}
|
|
|
|
dbConfig := &DBConfig{Path: fs.Arg(0)}
|
|
for _, u := range fs.Args()[1:] {
|
|
syncInterval := litestream.DefaultSyncInterval
|
|
dbConfig.Replicas = append(dbConfig.Replicas, &ReplicaConfig{
|
|
URL: u,
|
|
SyncInterval: &syncInterval,
|
|
})
|
|
}
|
|
c.Config.DBs = []*DBConfig{dbConfig}
|
|
} else {
|
|
if c.configPath == "" {
|
|
c.configPath = DefaultConfigPath()
|
|
}
|
|
if c.Config, err = ReadConfigFile(c.configPath, !c.noExpandEnv); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Override config with flags, if specified.
|
|
if *addr != "" {
|
|
c.Config.Addr = *addr
|
|
}
|
|
if *execFlag != "" {
|
|
c.Config.Exec = *execFlag
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Run loads all databases specified in the configuration.
|
|
func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
|
|
// Display version information.
|
|
log.Printf("litestream %s", Version)
|
|
|
|
// Setup databases.
|
|
if len(c.Config.DBs) == 0 {
|
|
log.Println("no databases specified in configuration")
|
|
}
|
|
|
|
c.server = litestream.NewServer()
|
|
if err := c.server.Open(); err != nil {
|
|
return fmt.Errorf("open server: %w", err)
|
|
}
|
|
|
|
// Add databases to the server.
|
|
for _, dbConfig := range c.Config.DBs {
|
|
path, err := expand(dbConfig.Path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := c.server.Watch(path, func(path string) (*litestream.DB, error) {
|
|
return NewDBFromConfigWithPath(dbConfig, path)
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Notify user that initialization is done.
|
|
for _, db := range c.server.DBs() {
|
|
log.Printf("initialized db: %s", db.Path())
|
|
for _, r := range db.Replicas {
|
|
switch client := r.Client().(type) {
|
|
case *litestream.FileReplicaClient:
|
|
log.Printf("replicating to: name=%q type=%q path=%q", r.Name(), client.Type(), client.Path())
|
|
case *s3.ReplicaClient:
|
|
log.Printf("replicating to: name=%q type=%q bucket=%q path=%q region=%q endpoint=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, client.Region, client.Endpoint, r.SyncInterval)
|
|
case *gs.ReplicaClient:
|
|
log.Printf("replicating to: name=%q type=%q bucket=%q path=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, r.SyncInterval)
|
|
case *abs.ReplicaClient:
|
|
log.Printf("replicating to: name=%q type=%q bucket=%q path=%q endpoint=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, client.Endpoint, r.SyncInterval)
|
|
case *sftp.ReplicaClient:
|
|
log.Printf("replicating to: name=%q type=%q host=%q user=%q path=%q sync-interval=%s", r.Name(), client.Type(), client.Host, client.User, client.Path, r.SyncInterval)
|
|
default:
|
|
log.Printf("replicating to: name=%q type=%q", r.Name(), client.Type())
|
|
}
|
|
}
|
|
}
|
|
|
|
// Serve HTTP if enabled.
|
|
if c.Config.Addr != "" {
|
|
c.httpServer = http.NewServer(c.server, c.Config.Addr)
|
|
if err := c.httpServer.Open(); err != nil {
|
|
return fmt.Errorf("cannot start http server: %w", err)
|
|
}
|
|
log.Printf("http server running at %s", c.httpServer.URL())
|
|
}
|
|
|
|
// Parse exec commands args & start subprocess.
|
|
if c.Config.Exec != "" {
|
|
execArgs, err := shellwords.Parse(c.Config.Exec)
|
|
if err != nil {
|
|
return fmt.Errorf("cannot parse exec command: %w", err)
|
|
}
|
|
|
|
// Pass first database path to child process.
|
|
env := os.Environ()
|
|
if dbs := c.server.DBs(); len(dbs) > 0 {
|
|
env = append(env, fmt.Sprintf("LITESTREAM_DB_PATH=%s", dbs[0].Path()))
|
|
}
|
|
|
|
c.cmd = exec.CommandContext(ctx, execArgs[0], execArgs[1:]...)
|
|
c.cmd.Env = env
|
|
c.cmd.Stdout = os.Stdout
|
|
c.cmd.Stderr = os.Stderr
|
|
if err := c.cmd.Start(); err != nil {
|
|
return fmt.Errorf("cannot start exec command: %w", err)
|
|
}
|
|
go func() { c.execCh <- c.cmd.Wait() }()
|
|
}
|
|
|
|
log.Printf("litestream initialization complete")
|
|
|
|
return nil
|
|
}
|
|
|
|
// Close closes the HTTP server & all open databases.
|
|
func (c *ReplicateCommand) Close() (err error) {
|
|
if c.httpServer != nil {
|
|
if e := c.httpServer.Close(); e != nil && err == nil {
|
|
err = e
|
|
}
|
|
}
|
|
if c.server != nil {
|
|
if e := c.server.Close(); e != nil && err == nil {
|
|
err = e
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
// Usage prints the help screen to STDOUT.
|
|
func (c *ReplicateCommand) Usage() {
|
|
fmt.Fprintf(c.stdout, `
|
|
The replicate command starts a server to monitor & replicate databases.
|
|
You can specify your database & replicas in a configuration file or you can
|
|
replicate a single database file by specifying its path and its replicas in the
|
|
command line arguments.
|
|
|
|
Usage:
|
|
|
|
litestream replicate [arguments]
|
|
|
|
litestream replicate [arguments] DB_PATH REPLICA_URL [REPLICA_URL...]
|
|
|
|
Arguments:
|
|
|
|
-config PATH
|
|
Specifies the configuration file.
|
|
Defaults to %s
|
|
|
|
-exec CMD
|
|
Executes a subcommand. Litestream will exit when the child
|
|
process exits. Useful for simple process management.
|
|
|
|
-addr BIND_ADDR
|
|
Starts an HTTP server that reports prometheus metrics and provides
|
|
an endpoint for live read replication. (e.g. ":9090")
|
|
|
|
-no-expand-env
|
|
Disables environment variable expansion in configuration file.
|
|
|
|
`[1:], DefaultConfigPath())
|
|
}
|