snapshots

Documentation

Index

• type Database
  • func NewDatabase(logger *zap.SugaredLogger, connection string) *Database
  • func (db *Database) AddSnapshotTags(ctx context.Context, id string, tags []string) error
  • func (db *Database) Close()
  • func (db *Database) CreateSnapshot(ctx context.Context, record *DatabaseRecord) (string, error)
  • func (db *Database) DeleteSnapshot(ctx context.Context, id string) error
  • func (db *Database) GetSnapshot(ctx context.Context, id string) (*DatabaseRecord, error)
  • func (db *Database) ListSnapshots(ctx context.Context, example *DatabaseRecord, pageSize int) ([]*DatabaseRecord, error)
  • func (db *Database) RemoveSnapshotTags(ctx context.Context, id string, tags []string) error
  • func (db *Database) SetSnapshotSize(ctx context.Context, id string, size uint64) error
  • func (db *Database) SetSnapshotTags(ctx context.Context, id string, tags []string) error
• type DatabaseRecord
• type Manager
  • func NewManager(op ManagerOptions) *Manager
  • func (m *Manager) Close()
• type ManagerOptions
• type Server
  • func (s *Server) AddSnapshotTags(ctx context.Context, request *snapshots_v1.AddSnapshotTagsRequest) (*snapshots_v1.Snapshot, error)
  • func (s *Server) CreateSnapshot(ctx context.Context, request *snapshots_v1.CreateSnapshotRequest) (*snapshots_v1.CreateSnapshotResponse, error)
  • func (s *Server) DeleteSnapshot(ctx context.Context, request *snapshots_v1.DeleteSnapshotRequest) (*emptypb.Empty, error)
  • func (s *Server) GetSnapshot(ctx context.Context, request *snapshots_v1.GetSnapshotRequest) (*snapshots_v1.GetSnapshotResponse, error)
  • func (s *Server) GetStorePath(_ context.Context, _ *emptypb.Empty) (*snapshots_v1.GetStorePathResponse, error)
  • func (s *Server) ListSnapshots(ctx context.Context, request *snapshots_v1.ListSnapshotsRequest) (*snapshots_v1.ListSnapshotsResponse, error)
  • func (s *Server) MarkReady(ctx context.Context, request *snapshots_v1.MarkReadyRequest) (*emptypb.Empty, error)
  • func (s *Server) RemoveSnapshotTags(ctx context.Context, request *snapshots_v1.RemoveSnapshotTagsRequest) (*snapshots_v1.Snapshot, error)
  • func (s *Server) SetSnapshotTags(ctx context.Context, request *snapshots_v1.SetSnapshotTagsRequest) (*snapshots_v1.Snapshot, error)
  • func (s *Server) StreamArchive(request *snapshots_v1.StreamArchiveRequest, ...) error
• type Store
  • func NewStore(logger *zap.SugaredLogger, root string) *Store
  • func (s *Store) CheckData(id string) string
  • func (s *Store) CreateTemporary(id string, expectedSize uint64) (string, error)
  • func (s *Store) DeleteSnapshot(id string)
  • func (s *Store) ExtractArchive(ctx context.Context, id string, r io.Reader) error
  • func (s *Store) GetNodeID() string
  • func (s *Store) PruneCache(maxSize uint64) (uint64, error)
  • func (s *Store) SaveTemporary(id string, isCache bool) (uint64, error)
  • func (s *Store) StreamArchive(ctx context.Context, id string) (io.ReadCloser, error)

Types

type Database

type Database struct {
	// contains filtered or unexported fields
}

func NewDatabase

func NewDatabase(logger *zap.SugaredLogger, connection string) *Database

func (*Database) AddSnapshotTags

func (db *Database) AddSnapshotTags(ctx context.Context, id string, tags []string) error

func (*Database) Close

func (db *Database) Close()

func (*Database) CreateSnapshot

func (db *Database) CreateSnapshot(ctx context.Context, record *DatabaseRecord) (string, error)

CreateSnapshot creates a new snapshot record in the database, generating a new ID and returns it.

func (*Database) DeleteSnapshot

func (db *Database) DeleteSnapshot(ctx context.Context, id string) error

func (*Database) GetSnapshot

func (db *Database) GetSnapshot(ctx context.Context, id string) (*DatabaseRecord, error)

func (*Database) ListSnapshots

func (db *Database) ListSnapshots(ctx context.Context, example *DatabaseRecord, pageSize int) ([]*DatabaseRecord, error)

func (*Database) RemoveSnapshotTags

func (db *Database) RemoveSnapshotTags(ctx context.Context, id string, tags []string) error

func (*Database) SetSnapshotSize

func (db *Database) SetSnapshotSize(ctx context.Context, id string, size uint64) error

func (*Database) SetSnapshotTags

func (db *Database) SetSnapshotTags(ctx context.Context, id string, tags []string) error

type DatabaseRecord

type DatabaseRecord struct {
	ID        uuid.UUID
	ParentID  uuid.NullUUID
	Hierarchy string
	TaskType  sql.NullString
	TaskName  sql.NullString
	TaskIndex types.NullTaskIndex
	EnvType   sql.NullString
	SessionID sql.NullInt64
	Step      sql.NullInt32
	Node      string
	Size      sql.NullInt64
	CreatedAt time.Time
	Tags      []string
}

type Manager

type Manager struct {
	Logger       *zap.SugaredLogger
	NodeRegistry *cluster.NodeRegistry
	Database     *Database
	Store        *Store
	Server       *Server
}

func NewManager

func NewManager(op ManagerOptions) *Manager

func (*Manager) Close

func (m *Manager) Close()

type ManagerOptions

type ManagerOptions struct {
	Logger             *zap.SugaredLogger
	NodeRegistry       *cluster.NodeRegistry
	HttpServer         *echo.Echo
	GrpcServer         *grpc.Server
	DatabaseConnection string
	StoreDirectory     string
}

type Server

type Server struct {
	snapshots_v1.UnimplementedSnapshotsManagerServer
	// contains filtered or unexported fields
}

func (*Server) AddSnapshotTags

func (s *Server) AddSnapshotTags(ctx context.Context, request *snapshots_v1.AddSnapshotTagsRequest) (*snapshots_v1.Snapshot, error)

func (*Server) CreateSnapshot

func (s *Server) CreateSnapshot(ctx context.Context, request *snapshots_v1.CreateSnapshotRequest) (*snapshots_v1.CreateSnapshotResponse, error)

func (*Server) DeleteSnapshot

func (s *Server) DeleteSnapshot(ctx context.Context, request *snapshots_v1.DeleteSnapshotRequest) (*emptypb.Empty, error)

func (*Server) GetSnapshot

func (s *Server) GetSnapshot(ctx context.Context, request *snapshots_v1.GetSnapshotRequest) (*snapshots_v1.GetSnapshotResponse, error)

func (*Server) GetStorePath

func (s *Server) GetStorePath(_ context.Context, _ *emptypb.Empty) (*snapshots_v1.GetStorePathResponse, error)

func (*Server) ListSnapshots

func (s *Server) ListSnapshots(ctx context.Context, request *snapshots_v1.ListSnapshotsRequest) (*snapshots_v1.ListSnapshotsResponse, error)

func (*Server) MarkReady

func (s *Server) MarkReady(ctx context.Context, request *snapshots_v1.MarkReadyRequest) (*emptypb.Empty, error)

func (*Server) RemoveSnapshotTags

func (s *Server) RemoveSnapshotTags(ctx context.Context, request *snapshots_v1.RemoveSnapshotTagsRequest) (*snapshots_v1.Snapshot, error)

func (*Server) SetSnapshotTags

func (s *Server) SetSnapshotTags(ctx context.Context, request *snapshots_v1.SetSnapshotTagsRequest) (*snapshots_v1.Snapshot, error)

func (*Server) StreamArchive

func (s *Server) StreamArchive(request *snapshots_v1.StreamArchiveRequest, stream grpc.ServerStreamingServer[snapshots_v1.ArchiveChunk]) error

type Store

type Store struct {
	RootPath      string
	DataPath      string
	CachePath     string
	TmpPath       string
	NodeIDPath    string
	CapacityLimit int // percentage of disk space to keep free
	// contains filtered or unexported fields
}

func NewStore

func NewStore(logger *zap.SugaredLogger, root string) *Store

func (*Store) CheckData

func (s *Store) CheckData(id string) string

CheckData checks if a snapshot with the given ID exists in the data or the cache. Return the path if it exists, otherwise returns an empty string.

func (*Store) CreateTemporary

func (s *Store) CreateTemporary(id string, expectedSize uint64) (string, error)

CreateTemporary creates a new temporary directory for building a snapshot. It checks if there is enough disk space available, considering the expected size of the snapshot. Returns the path to the temporary directory.

func (*Store) DeleteSnapshot

func (s *Store) DeleteSnapshot(id string)

DeleteSnapshot deletes a snapshot from data, cache, and temporary directories.

func (*Store) ExtractArchive

func (s *Store) ExtractArchive(ctx context.Context, id string, r io.Reader) error

ExtractArchive extracts a tar archive stream into a temporary snapshot directory. The temporary directory must be created with CreateTemporary first and be saving with SaveTemporary after use. The caller is responsible for closing the reader.

func (*Store) GetNodeID

func (s *Store) GetNodeID() string

GetNodeID get a locally persisted stable Node ID. Currently not used.

func (*Store) PruneCache

func (s *Store) PruneCache(maxSize uint64) (uint64, error)

PruneCache removes least recently used snapshots from the cache. If maxSize is provided, only remove no more than that many bytes of cache. Returns the total size of removed cache in bytes.

func (*Store) SaveTemporary

func (s *Store) SaveTemporary(id string, isCache bool) (uint64, error)

SaveTemporary atomically moves the temporary snapshot directory to the data directory. Returns the size of the saved snapshot in bytes.

func (*Store) StreamArchive

func (s *Store) StreamArchive(ctx context.Context, id string) (io.ReadCloser, error)

StreamArchive streams the snapshot directory as a tar archive. The caller is responsible for closing the returned ReadCloser.