waveterm/backup
0
0
Fork 0
mirror of https://github.com/wavetermdev/backup.git synced 2026-08-05 13:57:07 -07:00
Code Issues Packages Projects Releases Wiki Activity

write pty output to blockstore. initialize blockstore file on controller start. create frontend api to read the blockfile

Browse Source
This commit is contained in:
sawka
2024-05-28 21:44:47 -07:00
parent bc18869b2e
commit bff46d9822
4 changed files with 109 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.go
+45
View File
@@ -9,6 +9,7 @@ import (
"context"
"embed"
"fmt"
"io/fs"
"log"
"net/http"
"os"
@@ -18,6 +19,7 @@ import (
"syscall"
"time"
"github.com/google/uuid"
"github.com/wavetermdev/thenextwave/pkg/blockstore"
"github.com/wavetermdev/thenextwave/pkg/eventbus"
"github.com/wavetermdev/thenextwave/pkg/service/blockservice"
@@ -98,13 +100,56 @@ type waveAssetHandler struct {
AssetHandler http.Handler
}
func serveBlockFile(w http.ResponseWriter, r *http.Request) {
blockId := r.URL.Query().Get("blockid")
name := r.URL.Query().Get("name")
if _, err := uuid.Parse(blockId); err != nil {
http.Error(w, fmt.Sprintf("invalid blockid: %v", err), http.StatusBadRequest)
return
}
if name == "" {
http.Error(w, "name is required", http.StatusBadRequest)
return
}
file, err := blockstore.GBS.Stat(r.Context(), blockId, name)
if err == fs.ErrNotExist {
http.NotFound(w, r)
return
}
if err != nil {
http.Error(w, fmt.Sprintf("error getting file info: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/octet-stream")
w.Header().Set("Content-Length", fmt.Sprintf("%d", file.Size))
for offset := file.DataStartIdx(); offset < file.Size; offset += blockstore.DefaultPartDataSize {
_, data, err := blockstore.GBS.ReadAt(r.Context(), blockId, name, offset, blockstore.DefaultPartDataSize)
if err != nil {
if offset == 0 {
http.Error(w, fmt.Sprintf("error reading file: %v", err), http.StatusInternalServerError)
} else {
// nothing to do, the headers have already been sent
log.Printf("error reading file %s/%s @ %d: %v\n", blockId, name, offset, err)
}
return
}
w.Write(data)
}
}
func serveWaveUrls(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-cache")
if r.URL.Path == "/wave/stream-file" {
fileName := r.URL.Query().Get("path")
fileName = wavebase.ExpandHomeDir(fileName)
http.ServeFile(w, r, fileName)
return
}
if r.URL.Path == "/wave/blockfile" {
serveBlockFile(w, r)
return
}
http.NotFound(w, r)
}
pkg/blockcontroller/blockcontroller.go
+36 -9
View File
@@ -15,6 +15,7 @@ import (
"github.com/creack/pty"
"github.com/wailsapp/wails/v3/pkg/application"
"github.com/wavetermdev/thenextwave/pkg/blockstore"
"github.com/wavetermdev/thenextwave/pkg/eventbus"
"github.com/wavetermdev/thenextwave/pkg/shellexec"
"github.com/wavetermdev/thenextwave/pkg/wstore"
@@ -86,7 +87,35 @@ func (bc *BlockController) Close() {
}
}
const DefaultTermMaxFileSize = 256 * 1024
func (bc *BlockController) handleShellProcData(data []byte, seqNum int) error {
ctx, cancelFn := context.WithTimeout(context.Background(), DefaultTimeout)
defer cancelFn()
err := blockstore.GBS.AppendData(ctx, bc.BlockId, "main", data)
if err != nil {
return fmt.Errorf("error appending to blockfile: %w", err)
}
eventbus.SendEvent(application.WailsEvent{
Name: "block:ptydata",
Data: map[string]any{
"blockid": bc.BlockId,
"blockfile": "main",
"ptydata": base64.StdEncoding.EncodeToString(data),
"seqnum": seqNum,
},
})
return nil
}
func (bc *BlockController) DoRunShellCommand(rc *RunShellOpts) error {
// create a circular blockfile for the output
ctx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second)
defer cancelFn()
err := blockstore.GBS.MakeFile(ctx, bc.BlockId, "main", nil, blockstore.FileOptsType{MaxSize: DefaultTermMaxFileSize, Circular: true})
if err != nil && err != blockstore.ErrAlreadyExists {
return fmt.Errorf("error creating blockfile: %w", err)
}
if bc.getShellProc() != nil {
return nil
}
@@ -114,15 +143,13 @@ func (bc *BlockController) DoRunShellCommand(rc *RunShellOpts) error {
for {
nr, err := bc.ShellProc.Pty.Read(buf)
seqNum++
eventbus.SendEvent(application.WailsEvent{
Name: "block:ptydata",
Data: map[string]any{
"blockid": bc.BlockId,
"blockfile": "main",
"ptydata": base64.StdEncoding.EncodeToString(buf[:nr]),
"seqnum": seqNum,
},
})
if nr > 0 {
handleDataErr := bc.handleShellProcData(buf[:nr], seqNum)
if handleDataErr != nil {
log.Printf("error handling shell data: %v\n", handleDataErr)
break
}
}
if err == io.EOF {
break
}
pkg/blockstore/blockstore.go
+21 -3
View File
@@ -35,9 +35,9 @@ var GBS *BlockStore = &BlockStore{
}
type FileOptsType struct {
MaxSize int64
Circular bool
IJson bool
MaxSize int64 `json:"maxsize,omitempty"`
Circular bool `json:"circular,omitempty"`
IJson bool `json:"ijson,omitempty"`
}
type FileMeta = map[string]any
@@ -55,6 +55,24 @@ type BlockFile struct {
Meta FileMeta `json:"meta"` // only top-level keys can be updated (lower levels are immutable)
}
// for regular files this is just Size
// for circular files this is min(Size, MaxSize)
func (f BlockFile) DataLength() int64 {
if f.Opts.Circular {
return minInt64(f.Size, f.Opts.MaxSize)
}
return f.Size
}
// for regular files this is just 0
// for circular files this is the index of the first byte of data we have
func (f BlockFile) DataStartIdx() int64 {
if f.Opts.Circular && f.Size > f.Opts.MaxSize {
return f.Size - f.Opts.MaxSize
}
return 0
}
// this works because lower levels are immutable
func copyMeta(meta FileMeta) FileMeta {
newMeta := make(FileMeta)
pkg/blockstore/blockstore_dbops.go
+7 -1
View File
@@ -11,10 +11,16 @@ import (
"github.com/wavetermdev/waveterm/wavesrv/pkg/dbutil"
)
var ErrAlreadyExists = fmt.Errorf("file already exists")
func dbInsertFile(ctx context.Context, file *BlockFile) error {
// will fail if file already exists
return WithTx(ctx, func(tx *TxWrap) error {
query := "INSERT INTO db_block_file (blockid, name, size, createdts, modts, opts, meta) VALUES (?, ?, ?, ?, ?, ?, ?)"
query := "SELECT blockid FROM db_block_file WHERE blockid = ? AND name = ?"
if tx.Exists(query, file.BlockId, file.Name) {
return ErrAlreadyExists
}
query = "INSERT INTO db_block_file (blockid, name, size, createdts, modts, opts, meta) VALUES (?, ?, ?, ?, ?, ?, ?)"
tx.Exec(query, file.BlockId, file.Name, file.Size, file.CreatedTs, file.ModTs, dbutil.QuickJson(file.Opts), dbutil.QuickJson(file.Meta))
return nil
})