diff --git a/pkg/hostpath/hostpath.go b/pkg/hostpath/hostpath.go index c20c32c06..0f6ee3008 100644 --- a/pkg/hostpath/hostpath.go +++ b/pkg/hostpath/hostpath.go @@ -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 + if os.Getenv("CSI_NODE_ID") != "" && os.Getenv("CSI_ENDPOINT") != "" { + localStatefile = "/csi-publish/local-state.json" + } + + s, err := state.New(path.Join(cfg.StateDir, "state.json"), localStatefile, cfg.NodeID) if err != nil { return nil, err } diff --git a/pkg/state/state.go b/pkg/state/state.go index 967086c93..2c7de4a42 100644 --- a/pkg/state/state.go +++ b/pkg/state/state.go @@ -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 @@ -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 @@ -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{} @@ -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() @@ -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() + } + }() + err = unix.Flock(int(state.Fd()), unix.LOCK_EX) if err != nil { return status.Errorf(codes.Internal, "error locking state file: %v", err) @@ -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) + } + return nil } @@ -225,13 +257,35 @@ func (s *state) restore() error { } 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() + } + 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 } return nil @@ -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() } diff --git a/pkg/state/state_test.go b/pkg/state/state_test.go index 259f54d93..52f5ddfe8 100644 --- a/pkg/state/state_test.go +++ b/pkg/state/state_test.go @@ -30,8 +30,9 @@ 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") @@ -39,13 +40,15 @@ func TestVolumes(t *testing.T) { 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) + _, err = s.GetVolumeByName("bar") require.NoError(t, err, "get existing volume by name") @@ -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") @@ -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") @@ -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"}) @@ -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'") @@ -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") @@ -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")