77 "database/sql"
88 "fmt"
99 "os"
10+ "strings"
1011 "sync"
1112
1213 "github.com/k3s-io/kine/pkg/drivers"
@@ -39,16 +40,20 @@ var (
3940 `CREATE INDEX IF NOT EXISTS kine_id_deleted_index ON kine (id,deleted)` ,
4041 `CREATE INDEX IF NOT EXISTS kine_prev_revision_index ON kine (prev_revision)` ,
4142 `CREATE UNIQUE INDEX IF NOT EXISTS kine_name_prev_revision_uindex ON kine (name, prev_revision)` ,
42- `PRAGMA wal_checkpoint(TRUNCATE)` ,
4343 }
4444)
4545
4646func New (ctx context.Context , wg * sync.WaitGroup , cfg * drivers.Config ) (bool , server.Backend , error ) {
47- backend , _ , err := NewVariant (ctx , wg , "sqlite3" , cfg )
47+ backend , _ , err := NewVariant (ctx , wg , "sqlite3" , cfg , false )
4848 return false , backend , err
4949}
5050
51- func NewVariant (ctx context.Context , wg * sync.WaitGroup , driverName string , cfg * drivers.Config ) (server.Backend , * generic.Generic , error ) {
51+ func NewWithLitestream (ctx context.Context , wg * sync.WaitGroup , cfg * drivers.Config ) (bool , server.Backend , error ) {
52+ backend , _ , err := NewVariant (ctx , wg , "litestream" , cfg , true )
53+ return false , backend , err
54+ }
55+
56+ func NewVariant (ctx context.Context , wg * sync.WaitGroup , driverName string , cfg * drivers.Config , litestream bool ) (server.Backend , * generic.Generic , error ) {
5257 dataSourceName := cfg .DataSourceName
5358 if dataSourceName == "" {
5459 if err := os .MkdirAll ("./db" , 0700 ); err != nil {
@@ -57,6 +62,15 @@ func NewVariant(ctx context.Context, wg *sync.WaitGroup, driverName string, cfg
5762 dataSourceName = "./db/state.db?_journal=WAL&cache=shared&_busy_timeout=30000&_txlock=immediate"
5863 }
5964
65+ noCompactCheckpoint := strings .Contains (dataSourceName , "_kine_disable_compact_wal_checkpoint" )
66+ noAutoCheckpoint := strings .Contains (dataSourceName , "_kine_disable_wal_autocheckpoint" )
67+
68+ if driverName == "litestream" {
69+ logrus .Infof ("Litestream compatibility options enabled (all WAL checkpointing disabled)" )
70+ noCompactCheckpoint = true
71+ noAutoCheckpoint = true
72+ }
73+
6074 dialect , err := generic .Open (ctx , wg , driverName , dataSourceName , cfg .ConnectionPoolConfig , "?" , false , cfg .MetricsRegisterer )
6175 if err != nil {
6276 return nil , nil , err
@@ -81,7 +95,11 @@ func NewVariant(ctx context.Context, wg *sync.WaitGroup, driverName string, cfg
8195 kd.deleted != 0 AND
8296 kd.id <= ?
8397 )`
84- dialect .PostCompactSQL = `PRAGMA wal_checkpoint(FULL)`
98+ if noCompactCheckpoint {
99+ logrus .Infof ("WAL checkpoint on compact is disabled" )
100+ } else {
101+ dialect .PostCompactSQL = `PRAGMA wal_checkpoint(FULL)`
102+ }
85103 dialect .TranslateErr = func (err error ) error {
86104 if err , ok := err .(sqlite3.Error ); ok && err .ExtendedCode == sqlite3 .ErrConstraintUnique {
87105 return server .ErrKeyExists
@@ -98,17 +116,26 @@ func NewVariant(ctx context.Context, wg *sync.WaitGroup, driverName string, cfg
98116 return err .Error ()
99117 }
100118
101- if err := setup (dialect .DB ); err != nil {
119+ if err := setup (dialect .DB , noCompactCheckpoint , noAutoCheckpoint ); err != nil {
102120 return nil , nil , errors .Wrap (err , "setup db" )
103121 }
104122
105123 dialect .Migrate (context .Background ())
106124 return logstructured .New (sqllog .New (dialect , cfg .CompactInterval , cfg .CompactIntervalJitter , cfg .CompactTimeout , cfg .CompactMinRetain , cfg .CompactBatchSize , cfg .PollBatchSize )), dialect , nil
107125}
108126
109- func setup (db * sql.DB ) error {
127+ func setup (db * sql.DB , noCheckpointing , noAutoCheckpoint bool ) error {
110128 logrus .Infof ("Configuring database table schema and indexes, this may take a moment..." )
111129
130+ schema := append ([]string {}, schema ... )
131+ if ! noCheckpointing {
132+ schema = append (schema , `PRAGMA wal_checkpoint(TRUNCATE)` )
133+ }
134+ if noAutoCheckpoint {
135+ logrus .Infof ("WAL auto-checkpoint is disabled" )
136+ schema = append (schema , `PRAGMA wal_autocheckpoint = 0` )
137+ }
138+
112139 for _ , stmt := range schema {
113140 logrus .Tracef ("SETUP EXEC : %v" , util .Stripped (stmt ))
114141 _ , err := db .Exec (stmt )
@@ -122,6 +149,13 @@ func setup(db *sql.DB) error {
122149}
123150
124151func init () {
152+ sql .Register ("litestream" , & sqlite3.SQLiteDriver {
153+ ConnectHook : func (conn * sqlite3.SQLiteConn ) (err error ) {
154+ return conn .SetFileControlInt ("main" , sqlite3 .SQLITE_FCNTL_PERSIST_WAL , 1 )
155+ },
156+ })
157+
125158 drivers .Register ("sqlite" , New )
159+ drivers .Register ("litestream" , NewWithLitestream )
126160 drivers .SetDefault ("sqlite" )
127161}
0 commit comments