waveterm/backup
0
0
Fork 0
mirror of https://github.com/wavetermdev/backup.git synced 2026-04-22 15:26:58 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into evan/grid

Browse Source
This commit is contained in:
Evan Simkowitz
2024-05-06 14:18:25 -07:00
parent 1a9ba17a95 2772c76728
commit 45e2db0739
11 changed files with 247 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/regression.yml
+1 -1
View File
@@ -22,7 +22,7 @@ jobs:
- uses: dashcamio/testdriver@main
id: testdriver
with:
version: v2.10.2
version: v2.12.5
prerun: |
rm ~/Desktop/WITH-LOVE-FROM-AMERICA.txt
cd ~/actions-runner/_work/testdriver/testdriver/
wavesrv/cmd/main-server.go
+5
View File
@@ -1128,6 +1128,11 @@ func main() {
log.Printf("[error] migrate up: %v\n", err)
return
}
// err = blockstore.MigrateBlockstore()
// if err != nil {
// log.Printf("[error] migrate blockstore: %v\n", err)
// return
// }
clientData, err := sstore.EnsureClientData(context.Background())
if err != nil {
log.Printf("[error] ensuring client data: %v\n", err)
wavesrv/db/blockstore-migrations/000001_init.down.sql
+1
View File
@@ -0,0 +1 @@
-- nothing
wavesrv/db/blockstore-migrations/000001_init.up.sql
+19
View File
@@ -0,0 +1,19 @@
CREATE TABLE block_file (
blockid varchar(36) NOT NULL,
name varchar(200) NOT NULL,
maxsize bigint NOT NULL,
circular boolean NOT NULL,
size bigint NOT NULL,
createdts bigint NOT NULL,
modts bigint NOT NULL,
meta json NOT NULL,
PRIMARY KEY (blockid, name)
);
CREATE TABLE block_data (
blockid varchar(36) NOT NULL,
name varchar(200) NOT NULL,
partidx int NOT NULL,
data blob NOT NULL,
PRIMARY KEY(blockid, name, partidx)
);
wavesrv/db/db.go
+3
View File
@@ -10,3 +10,6 @@ import "embed"
//go:embed migrations/*.sql
var MigrationFS embed.FS
//go:embed blockstore-migrations/*.sql
var BlockstoreMigrationFS embed.FS
wavesrv/go.mod
-1
View File
@@ -5,7 +5,6 @@ go 1.22
toolchain go1.22.0
require (
github.com/alecthomas/units v0.0.0-20231202071711-9a357b53e9c9
github.com/alessio/shellescape v1.4.1
github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2
github.com/creack/pty v1.1.18
wavesrv/go.sum
-5
View File
@@ -1,5 +1,3 @@
github.com/alecthomas/units v0.0.0-20231202071711-9a357b53e9c9 h1:ez/4by2iGztzR4L0zgAOR8lTQK9VlyBVVd7G4omaOQs=
github.com/alecthomas/units v0.0.0-20231202071711-9a357b53e9c9/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
github.com/alessio/shellescape v1.4.1 h1:V7yhSDDn8LP4lc4jS8pFkt0zCnzVJlG5JXy9BVKJUX0=
github.com/alessio/shellescape v1.4.1/go.mod h1:PZAiSCk0LJaZkiCSkPv8qIobYglO3FPpyFjDCtHLS30=
github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2 h1:7Ip0wMmLHLRJdrloDxZfhMm0xrLXZS8+COSu2bXmEQs=
@@ -57,7 +55,6 @@ github.com/sawka/txwrap v0.1.2 h1:v8xS0Z1LE7/6vMZA81PYihI+0TSR6Zm1MalzzBIuXKc=
github.com/sawka/txwrap v0.1.2/go.mod h1:T3nlw2gVpuolo6/XEetvBbk1oMXnY978YmBFy1UyHvw=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/wavetermdev/ssh_config v0.0.0-20240306041034-17e2087ebde2 h1:onqZrJVap1sm15AiIGTfWzdr6cEF0KdtddeuuOVhzyY=
@@ -74,8 +71,6 @@ golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.15.0 h1:y/Oo/a/q3IXu26lQgl04j/gjuBDOBlx7X6Om1j2CPW4=
golang.org/x/term v0.15.0/go.mod h1:BDl952bC7+uMoWR75FIrCDx79TPU9oHkTZ9yRbYOrX0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
mvdan.cc/sh/v3 v3.7.0 h1:lSTjdP/1xsddtaKfGg7Myu7DnlHItd3/M2tomOcNNBg=
wavesrv/pkg/blockstore/blockstore.go
+37 -28
View File
@@ -10,8 +10,6 @@ import (
"strings"
"sync"
"time"
"github.com/alecthomas/units"
)
type FileOptsType struct {
@@ -32,7 +30,11 @@ type FileInfo struct {
Meta FileMeta
}
const MaxBlockSize = int64(128 * units.Kilobyte)
const UnitsKB = 1024 * 1024
const UnitsMB = 1024 * UnitsKB
const UnitsGB = 1024 * UnitsMB
const MaxBlockSize = int64(128 * UnitsKB)
const DefaultFlushTimeout = 1 * time.Second
type CacheEntry struct {
@@ -79,16 +81,23 @@ type BlockStore interface {
GetAllBlockIds(ctx context.Context) []string
}
var cache map[string]*CacheEntry = make(map[string]*CacheEntry)
var blockstoreCache map[string]*CacheEntry = make(map[string]*CacheEntry)
var globalLock *sync.Mutex = &sync.Mutex{}
var appendLock *sync.Mutex = &sync.Mutex{}
var flushTimeout = DefaultFlushTimeout
var lastWriteTime time.Time
// for testing
func clearCache() {
globalLock.Lock()
defer globalLock.Unlock()
blockstoreCache = make(map[string]*CacheEntry)
}
func InsertFileIntoDB(ctx context.Context, fileInfo FileInfo) error {
metaJson, err := json.Marshal(fileInfo.Meta)
if err != nil {
return fmt.Errorf("Error writing file %s to db: %v", fileInfo.Name, err)
return fmt.Errorf("error writing file %s to db: %v", fileInfo.Name, err)
}
txErr := WithTx(ctx, func(tx *TxWrap) error {
query := `INSERT INTO block_file VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
@@ -96,7 +105,7 @@ func InsertFileIntoDB(ctx context.Context, fileInfo FileInfo) error {
return nil
})
if txErr != nil {
return fmt.Errorf("Error writing file %s to db: %v", fileInfo.Name, txErr)
return fmt.Errorf("error writing file %s to db: %v", fileInfo.Name, txErr)
}
return nil
}
@@ -104,7 +113,7 @@ func InsertFileIntoDB(ctx context.Context, fileInfo FileInfo) error {
func WriteFileToDB(ctx context.Context, fileInfo FileInfo) error {
metaJson, err := json.Marshal(fileInfo.Meta)
if err != nil {
return fmt.Errorf("Error writing file %s to db: %v", fileInfo.Name, err)
return fmt.Errorf("error writing file %s to db: %v", fileInfo.Name, err)
}
txErr := WithTx(ctx, func(tx *TxWrap) error {
query := `UPDATE block_file SET blockid = ?, name = ?, maxsize = ?, circular = ?, size = ?, createdts = ?, modts = ?, meta = ? where blockid = ? and name = ?`
@@ -112,7 +121,7 @@ func WriteFileToDB(ctx context.Context, fileInfo FileInfo) error {
return nil
})
if txErr != nil {
return fmt.Errorf("Error writing file %s to db: %v", fileInfo.Name, txErr)
return fmt.Errorf("error writing file %s to db: %v", fileInfo.Name, txErr)
}
return nil
@@ -125,7 +134,7 @@ func WriteDataBlockToDB(ctx context.Context, blockId string, name string, index
return nil
})
if txErr != nil {
return fmt.Errorf("Error writing data block to db: %v", txErr)
return fmt.Errorf("error writing data block to db: %v", txErr)
}
return nil
}
@@ -152,7 +161,7 @@ func WriteToCacheBlockNum(ctx context.Context, blockId string, name string, p []
defer cacheEntry.Lock.Unlock()
block, err := GetCacheBlock(ctx, blockId, name, cacheNum, pullFromDB)
if err != nil {
return 0, 0, fmt.Errorf("Error getting cache block: %v", err)
return 0, 0, fmt.Errorf("error getting cache block: %v", err)
}
var bytesWritten = 0
blockLen := len(block.data)
@@ -192,7 +201,7 @@ func ReadFromCacheBlock(ctx context.Context, blockId string, name string, block
}
}()
if pos > len(block.data) {
return 0, fmt.Errorf("Reading past end of cache block, should never happen")
return 0, fmt.Errorf("reading past end of cache block, should never happen")
}
bytesWritten := 0
index := pos
@@ -216,7 +225,7 @@ func ReadFromCacheBlock(ctx context.Context, blockId string, name string, block
return bytesWritten, nil
}
const MaxSizeError = "Hit Max Size"
const MaxSizeError = "MaxSizeError"
func WriteToCacheBuf(buf *[]byte, p []byte, pos int, length int, maxWrite int64) (int, error) {
bytesToWrite := length
@@ -260,7 +269,7 @@ func GetValuesFromCacheId(cacheId string) (blockId string, name string) {
func GetCacheEntry(ctx context.Context, blockId string, name string) (*CacheEntry, bool) {
globalLock.Lock()
defer globalLock.Unlock()
if curCacheEntry, found := cache[GetCacheId(blockId, name)]; found {
if curCacheEntry, found := blockstoreCache[GetCacheId(blockId, name)]; found {
return curCacheEntry, true
} else {
return nil, false
@@ -279,7 +288,7 @@ func GetCacheEntryOrPopulate(ctx context.Context, blockId string, name string) (
if cacheEntry, found := GetCacheEntry(ctx, blockId, name); found {
return cacheEntry, nil
} else {
return nil, fmt.Errorf("Error getting cache entry %v %v", blockId, name)
return nil, fmt.Errorf("error getting cache entry %v %v", blockId, name)
}
}
@@ -288,16 +297,16 @@ func GetCacheEntryOrPopulate(ctx context.Context, blockId string, name string) (
func SetCacheEntry(ctx context.Context, cacheId string, cacheEntry *CacheEntry) {
globalLock.Lock()
defer globalLock.Unlock()
if _, found := cache[cacheId]; found {
if _, found := blockstoreCache[cacheId]; found {
return
}
cache[cacheId] = cacheEntry
blockstoreCache[cacheId] = cacheEntry
}
func DeleteCacheEntry(ctx context.Context, blockId string, name string) {
globalLock.Lock()
defer globalLock.Unlock()
delete(cache, GetCacheId(blockId, name))
delete(blockstoreCache, GetCacheId(blockId, name))
}
func GetCacheBlock(ctx context.Context, blockId string, name string, cacheNum int, pullFromDB bool) (*CacheBlock, error) {
@@ -392,7 +401,7 @@ func WriteAtHelper(ctx context.Context, blockId string, name string, p []byte, o
}
fInfo, err := Stat(ctx, blockId, name)
if err != nil {
return 0, fmt.Errorf("Write At err: %v", err)
return 0, fmt.Errorf("WriteAt err: %v", err)
}
if off > fInfo.Opts.MaxSize && fInfo.Opts.Circular {
numOver := off / fInfo.Opts.MaxSize
@@ -416,12 +425,12 @@ func WriteAtHelper(ctx context.Context, blockId string, name string, p []byte, o
b, err := WriteAtHelper(ctx, blockId, name, p, 0, false)
bytesWritten += b
if err != nil {
return bytesWritten, fmt.Errorf("Write to cache error: %v", err)
return bytesWritten, fmt.Errorf("write to cache error: %v", err)
}
break
}
} else {
return bytesWritten, fmt.Errorf("Write to cache error: %v", err)
return bytesWritten, fmt.Errorf("write to cache error: %v", err)
}
}
if len(p) == b {
@@ -452,7 +461,7 @@ func GetAllBlockSizes(dataBlocks []*CacheBlock) (int, int) {
}
func FlushCache(ctx context.Context) error {
for _, cacheEntry := range cache {
for _, cacheEntry := range blockstoreCache {
err := WriteFileToDB(ctx, *cacheEntry.Info)
if err != nil {
return err
@@ -485,14 +494,14 @@ func ReadAt(ctx context.Context, blockId string, name string, p *[]byte, off int
bytesRead := 0
fInfo, err := Stat(ctx, blockId, name)
if err != nil {
return 0, fmt.Errorf("Read At err: %v", err)
return 0, fmt.Errorf("ReadAt err: %v", err)
}
if off > fInfo.Opts.MaxSize && fInfo.Opts.Circular {
numOver := off / fInfo.Opts.MaxSize
off = off - (numOver * fInfo.Opts.MaxSize)
}
if off > fInfo.Size {
return 0, fmt.Errorf("Read At error: tried to read past the end of the file")
return 0, fmt.Errorf("ReadAt error: tried to read past the end of the file")
}
endReadPos := math.Min(float64(int64(len(*p))+off), float64(fInfo.Size))
bytesToRead := int64(endReadPos) - off
@@ -505,7 +514,7 @@ func ReadAt(ctx context.Context, blockId string, name string, p *[]byte, off int
for index := curCacheNum; index < curCacheNum+numCaches; index++ {
curCacheBlock, err := GetCacheBlock(ctx, blockId, name, index, true)
if err != nil {
return bytesRead, fmt.Errorf("Error getting cache block: %v", err)
return bytesRead, fmt.Errorf("error getting cache block: %v", err)
}
cacheOffset := off - (int64(index) * MaxBlockSize)
if cacheOffset < 0 {
@@ -540,7 +549,7 @@ func ReadAt(ctx context.Context, blockId string, name string, p *[]byte, off int
break
}
} else {
return bytesRead, fmt.Errorf("Read from cache error: %v", err)
return bytesRead, fmt.Errorf("read from cache error: %v", err)
}
}
}
@@ -552,7 +561,7 @@ func AppendData(ctx context.Context, blockId string, name string, p []byte) (int
defer appendLock.Unlock()
fInfo, err := Stat(ctx, blockId, name)
if err != nil {
return 0, fmt.Errorf("Append stat error: %v", err)
return 0, fmt.Errorf("append stat error: %v", err)
}
return WriteAt(ctx, blockId, name, p, fInfo.Size)
}
@@ -564,12 +573,12 @@ func DeleteFile(ctx context.Context, blockId string, name string) error {
}
func DeleteBlock(ctx context.Context, blockId string) error {
for cacheId, _ := range cache {
for cacheId := range blockstoreCache {
curBlockId, name := GetValuesFromCacheId(cacheId)
if curBlockId == blockId {
err := DeleteFile(ctx, blockId, name)
if err != nil {
return fmt.Errorf("Error deleting %v %v: %v", blockId, name, err)
return fmt.Errorf("error deleting %v %v: %v", blockId, name, err)
}
}
}
wavesrv/pkg/blockstore/blockstore_dbops.go
+64 -3
View File
@@ -8,11 +8,16 @@ import (
"path"
"sync"
"github.com/golang-migrate/migrate/v4"
_ "github.com/golang-migrate/migrate/v4/database/sqlite3"
"github.com/golang-migrate/migrate/v4/source/iofs"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
"github.com/sawka/txwrap"
"github.com/wavetermdev/waveterm/wavesrv/pkg/dbutil"
"github.com/wavetermdev/waveterm/wavesrv/pkg/scbase"
dbfs "github.com/wavetermdev/waveterm/wavesrv/db"
)
const DBFileName = "blockstore.db"
@@ -21,12 +26,64 @@ type SingleConnDBGetter struct {
SingleConnLock *sync.Mutex
}
var dbWrap *SingleConnDBGetter
var dbWrap *SingleConnDBGetter = &SingleConnDBGetter{SingleConnLock: &sync.Mutex{}}
type TxWrap = txwrap.TxWrap
func InitDBState() {
dbWrap = &SingleConnDBGetter{SingleConnLock: &sync.Mutex{}}
func MakeBlockstoreMigrate() (*migrate.Migrate, error) {
fsVar, err := iofs.New(dbfs.BlockstoreMigrationFS, "blockstore-migrations")
if err != nil {
return nil, fmt.Errorf("opening iofs: %w", err)
}
dbUrl := fmt.Sprintf("sqlite3://%s", GetDBName())
m, err := migrate.NewWithSourceInstance("iofs", fsVar, dbUrl)
if err != nil {
return nil, fmt.Errorf("making blockstore migration db[%s]: %w", GetDBName(), err)
}
return m, nil
}
func MigrateBlockstore() error {
log.Printf("migrate blockstore\n")
m, err := MakeBlockstoreMigrate()
if err != nil {
return err
}
curVersion, dirty, err := GetMigrateVersion(m)
if dirty {
return fmt.Errorf("cannot migrate up, database is dirty")
}
if err != nil {
return fmt.Errorf("cannot get current migration version: %v", err)
}
defer m.Close()
err = m.Up()
if err != nil && err != migrate.ErrNoChange {
return fmt.Errorf("migrating blockstore: %w", err)
}
newVersion, _, err := GetMigrateVersion(m)
if err != nil {
return fmt.Errorf("cannot get new migration version: %v", err)
}
if newVersion != curVersion {
log.Printf("[db] blockstore migration done, version %d -> %d\n", curVersion, newVersion)
}
return nil
}
func GetMigrateVersion(m *migrate.Migrate) (uint, bool, error) {
if m == nil {
var err error
m, err = MakeBlockstoreMigrate()
if err != nil {
return 0, false, err
}
}
curVersion, dirty, err := m.Version()
if err == migrate.ErrNilVersion {
return 0, false, nil
}
return curVersion, dirty, err
}
func (dbg *SingleConnDBGetter) GetDB(ctx context.Context) (*sqlx.DB, error) {
@@ -62,8 +119,12 @@ func WithTxRtn[RT any](ctx context.Context, fn func(tx *TxWrap) (RT, error)) (RT
var globalDBLock = &sync.Mutex{}
var globalDB *sqlx.DB
var globalDBErr error
var overrideDBName string
func GetDBName() string {
if overrideDBName != "" {
return overrideDBName
}
scHome := scbase.GetWaveHomeDir()
return path.Join(scHome, DBFileName)
}
wavesrv/pkg/blockstore/blockstore_test.go
+115 -97
View File
File diff suppressed because it is too large Load Diff
wavesrv/pkg/sstore/migrate.go
+2 -2
View File
@@ -17,7 +17,7 @@ import (
_ "github.com/golang-migrate/migrate/v4/source/file"
"github.com/golang-migrate/migrate/v4/source/iofs"
_ "github.com/mattn/go-sqlite3"
sh2db "github.com/wavetermdev/waveterm/wavesrv/db"
dbfs "github.com/wavetermdev/waveterm/wavesrv/db"
"github.com/golang-migrate/migrate/v4"
)
@@ -29,7 +29,7 @@ const CmdLineSpecialMigration = 20
const RISpecialMigration = 30
func MakeMigrate() (*migrate.Migrate, error) {
fsVar, err := iofs.New(sh2db.MigrationFS, "migrations")
fsVar, err := iofs.New(dbfs.MigrationFS, "migrations")
if err != nil {
return nil, fmt.Errorf("opening iofs: %w", err)
}