Skip to content

Move stage/publish volumes state to a dedicated local file #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cclerget merged 1 commit into from
Dec 16, 2025
Merged

Move stage/publish volumes state to a dedicated local file #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion pkg/hostpath/hostpath.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,13 @@ func NewHostPathDriver(cfg Config) (*hostPath, error) {
klog.Infof("Driver: %v ", cfg.DriverName)
klog.Infof("Version: %s", cfg.VendorVersion)

s, err := state.New(path.Join(cfg.StateDir, "state.json"))
localStatefile := path.Join(cfg.StateDir, "local-state.json")
// specific to fuzzball environment variables
Comment thread
cclerget marked this conversation as resolved.
if os.Getenv("CSI_NODE_ID") != "" && os.Getenv("CSI_ENDPOINT") != "" {
localStatefile = "/csi-publish/local-state.json"
}
Comment thread
cclerget marked this conversation as resolved.

s, err := state.New(path.Join(cfg.StateDir, "state.json"), localStatefile, cfg.NodeID)
if err != nil {
return nil, err
}
Expand Down
89 changes: 73 additions & 16 deletions pkg/state/state.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,17 +41,7 @@ const (
BlockSizeBytes = 4096
)

type Volume struct {
VolName string
VolID string
VolSize int64
VolPath string
VolAccessType AccessType
ParentVolID string
ParentSnapID string
Ephemeral bool
NodeID string
Kind string
type LocalVolume struct {
ReadOnlyAttach bool
Attached bool
// Staged contains the staging target path at which the volume
Expand All @@ -63,6 +53,20 @@ type Volume struct {
Published Strings
}

type Volume struct {
VolName string
VolID string
VolSize int64
VolPath string
VolAccessType AccessType
ParentVolID string
ParentSnapID string
Ephemeral bool
NodeID string
Kind string
LocalVolume
}

type Snapshot struct {
Name string
Id string
Expand Down Expand Up @@ -156,11 +160,16 @@ type resources struct {
GroupSnapshots []GroupSnapshot
}

type localResources map[string]LocalVolume

type state struct {
resources
localResources

statefilePath string
modTime time.Time
statefilePath string
modTime time.Time
localStatefilePath string
nodeID string
}

var _ State = &state{}
Expand All @@ -169,9 +178,12 @@ var _ State = &state{}
// and then ensures that all changes are mirrored immediately in the
// given file. If not given, the initial state is empty and changes
// are not saved.
func New(statefilePath string) (State, error) {
func New(statefilePath string, localStatefilePath string, nodeID string) (State, error) {
s := &state{
statefilePath: statefilePath,
statefilePath: statefilePath,
localStatefilePath: localStatefilePath,
nodeID: nodeID,
localResources: make(localResources),
}

return s, s.restore()
Expand All @@ -190,6 +202,18 @@ func (s *state) dump() (errFn error) {
}
}()

localState, err := os.OpenFile(s.localStatefilePath, os.O_WRONLY|os.O_CREATE, 0600)
if err != nil {
return status.Errorf(codes.Internal, "error opening local state file: %v", err)
}
defer func() {
if errFn == nil {
errFn = localState.Close()
} else {
_ = localState.Close()
}
}()
Comment thread
cclerget marked this conversation as resolved.

err = unix.Flock(int(state.Fd()), unix.LOCK_EX)
if err != nil {
return status.Errorf(codes.Internal, "error locking state file: %v", err)
Expand All @@ -207,6 +231,14 @@ func (s *state) dump() (errFn error) {
return status.Errorf(codes.Internal, "error encoding volumes and snapshots: %v", err)
}

if err := localState.Truncate(0); err != nil {
return status.Errorf(codes.Internal, "error truncating local state file: %v", err)
}

if err := json.NewEncoder(localState).Encode(&s.localResources); err != nil {
return status.Errorf(codes.Internal, "error encoding local resources: %v", err)
}
Comment thread
cclerget marked this conversation as resolved.
Comment thread
cclerget marked this conversation as resolved.

return nil
}

Expand All @@ -225,13 +257,35 @@ func (s *state) restore() error {
}
Comment thread
cclerget marked this conversation as resolved.
defer state.Close()

localState, err := os.OpenFile(s.localStatefilePath, os.O_RDONLY, 0600)
if err != nil {
if !errors.Is(err, os.ErrNotExist) {
return status.Errorf(codes.Internal, "error reading local state file: %v", err)
}
} else {
defer localState.Close()
Comment thread
cclerget marked this conversation as resolved.
}

if err := unix.Flock(int(state.Fd()), unix.LOCK_SH); err != nil {
return status.Errorf(codes.Internal, "error locking state file: %v", err)
}
defer unix.Flock(int(state.Fd()), unix.LOCK_UN)

if err := json.NewDecoder(state).Decode(&s.resources); err != nil {
return status.Errorf(codes.Internal, "error encoding volumes and snapshots from state file %q: %v", s.statefilePath, err)
return status.Errorf(codes.Internal, "error decoding volumes and snapshots from state file %q: %v", s.statefilePath, err)
}

if localState != nil {
if err := json.NewDecoder(localState).Decode(&s.localResources); err != nil {
return status.Errorf(codes.Internal, "error decoding local resources from state file %q: %v", s.localStatefilePath, err)
}
}

for i, vol := range s.Volumes {
if localVolume, ok := s.localResources[vol.VolID]; ok {
s.Volumes[i].LocalVolume = localVolume
}
s.Volumes[i].NodeID = s.nodeID
Comment thread
cclerget marked this conversation as resolved.
}

return nil
Expand Down Expand Up @@ -300,16 +354,19 @@ func (s *state) UpdateVolume(update Volume) error {
for i, volume := range s.Volumes {
if volume.VolID == update.VolID {
s.Volumes[i] = update
s.localResources[update.VolID] = update.LocalVolume
return s.dump()
}
}
s.Volumes = append(s.Volumes, update)
s.localResources[update.VolID] = update.LocalVolume
return s.dump()
}

func (s *state) DeleteVolume(volID string) error {
for i, volume := range s.Volumes {
if volume.VolID == volID {
delete(s.localResources, volID)
s.Volumes = append(s.Volumes[:i], s.Volumes[i+1:]...)
return s.dump()
}
Expand Down
26 changes: 16 additions & 10 deletions pkg/state/state_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,22 +30,25 @@ var grpcError = status.Errorf(codes.NotFound, "not found")
func TestVolumes(t *testing.T) {
tmp := t.TempDir()
statefileName := path.Join(tmp, "state.json")
localStatefileName := path.Join(tmp, "local-state.json")

s, err := New(statefileName)
s, err := New(statefileName, localStatefileName, "node1")
require.NoError(t, err, "construct state")
require.Empty(t, s.GetVolumes(), "initial volumes")

_, err = s.GetVolumeByID("foo")
require.Equal(t, codes.NotFound, status.Convert(err).Code(), "GetVolumeByID of non-existent volume")
require.Contains(t, status.Convert(err).Message(), "foo")

err = s.UpdateVolume(Volume{VolID: "foo", VolName: "bar"})
err = s.UpdateVolume(Volume{VolID: "foo", VolName: "bar", LocalVolume: LocalVolume{Attached: true}})
require.NoError(t, err, "add volume")

s, err = New(statefileName)
s, err = New(statefileName, localStatefileName, "node1")
require.NoError(t, err, "reconstruct state")
_, err = s.GetVolumeByID("foo")
vol, err := s.GetVolumeByID("foo")
require.NoError(t, err, "get existing volume by ID")
require.Equal(t, true, vol.Attached)
Comment thread
cclerget marked this conversation as resolved.
Comment thread
cclerget marked this conversation as resolved.

_, err = s.GetVolumeByName("bar")
require.NoError(t, err, "get existing volume by name")

Expand All @@ -61,8 +64,9 @@ func TestVolumes(t *testing.T) {
func TestSnapshots(t *testing.T) {
tmp := t.TempDir()
statefileName := path.Join(tmp, "state.json")
localStatefileName := path.Join(tmp, "local-state.json")

s, err := New(statefileName)
s, err := New(statefileName, localStatefileName, "node1")
require.NoError(t, err, "construct state")
require.Empty(t, s.GetSnapshots(), "initial snapshots")

Expand All @@ -73,7 +77,7 @@ func TestSnapshots(t *testing.T) {
err = s.UpdateSnapshot(Snapshot{Id: "foo", Name: "bar"})
require.NoError(t, err, "add snapshot")

s, err = New(statefileName)
s, err = New(statefileName, localStatefileName, "node1")
require.NoError(t, err, "reconstruct state")
_, err = s.GetSnapshotByID("foo")
require.NoError(t, err, "get existing snapshot by ID")
Expand All @@ -94,8 +98,9 @@ func TestSnapshots(t *testing.T) {
func TestSnapshotsFromSameSource(t *testing.T) {
tmp := t.TempDir()
statefileName := path.Join(tmp, "state.json")
localStatefileName := path.Join(tmp, "local-state.json")

s, err := New(statefileName)
s, err := New(statefileName, localStatefileName, "node1")
require.NoError(t, err, "construct state")

err = s.UpdateSnapshot(Snapshot{Id: "foo", Name: "foo-name", VolID: "source"})
Expand All @@ -113,7 +118,7 @@ func TestSnapshotsFromSameSource(t *testing.T) {
require.NoError(t, err, "get existing snapshot by name 'bar-name'")

// Make sure it still works after reconstruction
s, err = New(statefileName)
s, err = New(statefileName, localStatefileName, "node1")
require.NoError(t, err, "reconstruct state")
_, err = s.GetSnapshotByID("foo")
require.NoError(t, err, "get existing snapshot by ID 'foo'")
Expand All @@ -128,8 +133,9 @@ func TestSnapshotsFromSameSource(t *testing.T) {
func TestVolumeGroupSnapshots(t *testing.T) {
tmp := t.TempDir()
statefileName := path.Join(tmp, "state.json")
localStatefileName := path.Join(tmp, "local-state.json")

s, err := New(statefileName)
s, err := New(statefileName, localStatefileName, "node1")
require.NoError(t, err, "construct state")
require.Empty(t, s.GetGroupSnapshots(), "initial groupsnapshots")

Expand All @@ -140,7 +146,7 @@ func TestVolumeGroupSnapshots(t *testing.T) {
err = s.UpdateGroupSnapshot(GroupSnapshot{Id: "foo", Name: "bar"})
require.NoError(t, err, "add groupsnapshot")

s, err = New(statefileName)
s, err = New(statefileName, localStatefileName, "node1")
require.NoError(t, err, "reconstruct state")
_, err = s.GetGroupSnapshotByID("foo")
require.NoError(t, err, "get existing groupsnapshot by ID")
Expand Down