-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(CSI-308): implement database and sqlite3 backing
- Loading branch information
1 parent
66f8205
commit b57fee4
Showing
5 changed files
with
343 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,184 @@ | ||
| package db | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "github.com/container-storage-interface/spec/lib/go/csi" | ||
| "github.com/rs/zerolog/log" | ||
| "gorm.io/driver/sqlite" | ||
| "gorm.io/gorm" | ||
| "os" | ||
| "path/filepath" | ||
| ) | ||
|
|
||
| const DBPath = "/tmp/csi-wekafs-attachments/csi-attachments.db" | ||
|
|
||
| type PvcAttachment struct { | ||
| ID uint `gorm:"primaryKey"` // Auto-incrementing ID | ||
| VolumeId string `gorm:"index:idx_volume_target,unique"` | ||
| TargetPath string `gorm:"index:idx_volume_target,unique"` | ||
| Node string `json:"node"` | ||
| BootID string `json:"boot_id"` | ||
| AccessType string `json:"access_type"` | ||
| } | ||
|
|
||
| func (pal *PvcAttachment) String() string { | ||
| return fmt.Sprintf("PVC: %s, Node: %s, TargetPath: %s, BootID: %s, AccessType: %s", pal.VolumeId, pal.Node, pal.TargetPath, pal.BootID, pal.AccessType) | ||
| } | ||
|
|
||
| func (pal *PvcAttachment) MatchesBootId(bootID string) bool { | ||
| return pal.BootID == bootID | ||
| } | ||
|
|
||
| func (pal *PvcAttachment) MatchesNode(node string) bool { | ||
| return pal.Node == node | ||
| } | ||
|
|
||
| func (pal *PvcAttachment) MatchesVolumeId(volumeId string) bool { | ||
| return pal.VolumeId == volumeId | ||
| } | ||
|
|
||
| func (pal *PvcAttachment) MatchesAccessType(accessType string) bool { | ||
| return pal.AccessType == accessType | ||
| } | ||
|
|
||
| func (pal *PvcAttachment) MatchesTargetPath(path string) bool { | ||
| return pal.TargetPath == path | ||
| } | ||
|
|
||
| func (pal *PvcAttachment) Matches(volumeId, path, node, accessType string, bootId string) bool { | ||
| return pal.MatchesVolumeId(volumeId) && pal.MatchesTargetPath(path) && pal.MatchesNode(node) && pal.MatchesBootId(bootId) && pal.MatchesAccessType(accessType) | ||
| } | ||
|
|
||
| func (pal *PvcAttachment) IsSingleWriter() bool { | ||
| return pal.AccessType == csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER.String() || | ||
| pal.AccessType == csi.VolumeCapability_AccessMode_SINGLE_NODE_SINGLE_WRITER.String() | ||
| } | ||
|
|
||
| // GetDatabase returns a database for pod attachments that will be used on each node to satisfy the ReadWriteOncePod attachment mode | ||
| func GetDatabase(ctx context.Context) (Database, error) { | ||
| logger := log.Ctx(ctx) | ||
| directory := filepath.Dir(DBPath) | ||
| if err := EnsureDirectoryExists(directory); err != nil { | ||
| logger.Error().Err(err).Msg("Failed to create directory") | ||
| return nil, err | ||
| } | ||
| db, err := gorm.Open(sqlite.Open(DBPath), &gorm.Config{}) | ||
| if err != nil { | ||
| logger.Error().Err(err).Msg("Failed to connect to the database") | ||
| } | ||
|
|
||
| // Auto-migrate the schema | ||
| if err := db.AutoMigrate(&PvcAttachment{}); err != nil { | ||
| logger.Error().Err(err).Msg("Failed to migrate the database") | ||
| } | ||
|
|
||
| return &SqliteDatabase{ | ||
| db, | ||
| }, nil | ||
| } | ||
|
|
||
| func EnsureDirectoryExists(directory string) error { | ||
| _, err := os.Stat(directory) | ||
| if err != nil { | ||
| if os.IsNotExist(err) { | ||
| if err := os.MkdirAll(directory, 0755); err != nil { | ||
| return fmt.Errorf("failed to create directory: %w", err) | ||
| } | ||
| } else { | ||
| return fmt.Errorf("failed to stat directory: %w", err) | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| type SqliteDatabase struct { | ||
| *gorm.DB | ||
| } | ||
|
|
||
| func (d *SqliteDatabase) GetAttachmentsByVolumeIdOrTargetPath(ctx context.Context, volumeId, targetPath *string) (*[]PvcAttachment, error) { | ||
| if d == nil { | ||
| return nil, errors.New("database is nil") | ||
| } | ||
| query := d.Model(&PvcAttachment{}) | ||
| if volumeId != nil { | ||
| query = query.Where("volume_id = ?", *volumeId) | ||
| } | ||
| if targetPath != nil { | ||
| query = query.Where("target_path = ?", *targetPath) | ||
| } | ||
| locks := &[]PvcAttachment{} | ||
|
|
||
| err := query.Find(locks).Error | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to lookup records: %w", err) | ||
| } | ||
|
|
||
| return locks, nil | ||
| } | ||
|
|
||
| func (d *SqliteDatabase) CreateAttachment(ctx context.Context, attachment *PvcAttachment) error { | ||
| if d == nil { | ||
| return errors.New("database is nil") | ||
| } | ||
| if err := d.Create(attachment).Error; err != nil { | ||
| return fmt.Errorf("failed to create record: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (d *SqliteDatabase) UpdateAttachment(ctx context.Context, attachment *PvcAttachment) error { | ||
| if d == nil { | ||
| return errors.New("database is nil") | ||
| } | ||
| existing, err := d.GetAttachmentsByVolumeIdOrTargetPath(ctx, &attachment.VolumeId, &attachment.TargetPath) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to lookup existing record: %w", err) | ||
| } | ||
| if len(*existing) == 0 { | ||
| return errors.New("no record found") | ||
| } | ||
| attachment.ID = (*existing)[0].ID | ||
|
|
||
| if err := d.Save(attachment).Error; err != nil { | ||
| return fmt.Errorf("failed to update record: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (d *SqliteDatabase) CreateOrUpdateAttachment(ctx context.Context, attachment *PvcAttachment) error { | ||
| if d == nil { | ||
| return errors.New("database is nil") | ||
| } | ||
| existing, err := d.GetAttachmentsByVolumeIdOrTargetPath(ctx, &attachment.VolumeId, &attachment.TargetPath) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to lookup existing record: %w", err) | ||
| } | ||
| if len(*existing) == 0 { | ||
| return d.CreateAttachment(ctx, attachment) | ||
| } | ||
| return d.UpdateAttachment(ctx, attachment) | ||
| } | ||
|
|
||
| func (d *SqliteDatabase) DeleteAttachment(ctx context.Context, attachment *PvcAttachment) error { | ||
| if d == nil { | ||
| return errors.New("database is nil") | ||
| } | ||
| if err := d.Delete(attachment).Error; err != nil { | ||
| return fmt.Errorf("failed to delete record: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (d *SqliteDatabase) DeleteAttachmentDisregardingAccessType(volumeId, targetPath, node, bootId string) error { | ||
| // Build the delete query | ||
| result := d.Where("volume_id = ? AND target_path = ? AND node = ? AND boot_id = ?", volumeId, targetPath, node, bootId).Delete(&PvcAttachment{}) | ||
| if result.Error != nil { | ||
| return fmt.Errorf("failed to delete record: %w", result.Error) | ||
| } | ||
| if result.RowsAffected == 0 { | ||
| return fmt.Errorf("no record found for VolumeId: %s and TargetPath: %s", volumeId, targetPath) | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| package db | ||
|
|
||
| import ( | ||
| "context" | ||
| "os" | ||
| "testing" | ||
|
|
||
| "github.com/google/uuid" | ||
| "github.com/stretchr/testify/assert" | ||
| "gorm.io/driver/sqlite" | ||
| "gorm.io/gorm" | ||
| ) | ||
|
|
||
| func setupTestDB(t *testing.T) *SqliteDatabase { | ||
| _ = os.Remove(DBPath) | ||
| db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) | ||
| assert.NoError(t, err) | ||
|
|
||
| err = db.AutoMigrate(&PvcAttachment{}) | ||
| assert.NoError(t, err) | ||
|
|
||
| return &SqliteDatabase{db} | ||
| } | ||
|
|
||
| func TestDatabaseWrapper_GetAttachmentsByVolumeIdOrTargetPath(t *testing.T) { | ||
| db := setupTestDB(t) | ||
| ctx := context.Background() | ||
|
|
||
| volumeId := uuid.New().String() | ||
| targetPath := "/test/path" | ||
| attachment := &PvcAttachment{ | ||
| VolumeId: volumeId, | ||
| TargetPath: targetPath, | ||
| Node: "node1", | ||
| BootID: "boot1", | ||
| AccessType: "ReadWriteOnce", | ||
| } | ||
|
|
||
| // same attachment exactly | ||
| attachment2 := &PvcAttachment{ | ||
| VolumeId: volumeId, | ||
| TargetPath: targetPath, | ||
| Node: "node1", | ||
| BootID: "boot1", | ||
| AccessType: "ReadWriteOnce", | ||
| } | ||
|
|
||
| // attachment with different accesstype | ||
| attachment3 := &(*attachment2) | ||
| attachment3.AccessType = "ReadOnlyMany" | ||
|
|
||
| // attachment with different ID but same other values, should fail on unique constraints of volumeId and targetPath | ||
| attachment4 := &(*attachment3) | ||
| attachment4.ID = 10 | ||
|
|
||
| err := db.CreateAttachment(ctx, attachment) | ||
| assert.NoError(t, err) | ||
|
|
||
| err = db.UpdateAttachment(ctx, attachment2) | ||
| assert.NoError(t, err) | ||
|
|
||
| err = db.UpdateAttachment(ctx, attachment3) | ||
| assert.NoError(t, err) | ||
|
|
||
| err = db.CreateOrUpdateAttachment(ctx, attachment3) | ||
| assert.NoError(t, err) | ||
|
|
||
| err = db.CreateAttachment(ctx, attachment4) | ||
| assert.Error(t, err) | ||
|
|
||
| // Test by volumeId | ||
| attachments, err := db.GetAttachmentsByVolumeIdOrTargetPath(ctx, &volumeId, nil) | ||
| assert.NoError(t, err) | ||
| assert.Len(t, *attachments, 1) | ||
| assert.Equal(t, *attachment3, (*attachments)[0]) | ||
|
|
||
| // Test by targetPath | ||
| attachments, err = db.GetAttachmentsByVolumeIdOrTargetPath(ctx, nil, &targetPath) | ||
| assert.NoError(t, err) | ||
| assert.Len(t, *attachments, 1) | ||
| assert.Equal(t, *attachment3, (*attachments)[0]) | ||
| } | ||
|
|
||
| func TestDatabaseWrapper_DeleteAttachment(t *testing.T) { | ||
| db := setupTestDB(t) | ||
| ctx := context.Background() | ||
|
|
||
| attachment := &PvcAttachment{ | ||
| VolumeId: uuid.New().String(), | ||
| TargetPath: "/test/path", | ||
| Node: "node1", | ||
| BootID: "boot1", | ||
| AccessType: "ReadWriteOnce", | ||
| } | ||
|
|
||
| err := db.CreateOrUpdateAttachment(ctx, attachment) | ||
| assert.NoError(t, err) | ||
|
|
||
| err = db.DeleteAttachment(ctx, attachment) | ||
| assert.NoError(t, err) | ||
|
|
||
| var result PvcAttachment | ||
| err = db.First(&result, "volume_id = ?", attachment.VolumeId).Error | ||
| assert.Error(t, err) | ||
| assert.Equal(t, gorm.ErrRecordNotFound, err) | ||
| } | ||
|
|
||
| func TestDatabaseWrapper_DeleteAttachmentDisregardingAccessType(t *testing.T) { | ||
| db := setupTestDB(t) | ||
| ctx := context.Background() | ||
|
|
||
| attachment := &PvcAttachment{ | ||
| VolumeId: uuid.New().String(), | ||
| TargetPath: "/test/path", | ||
| Node: "node1", | ||
| BootID: "boot1", | ||
| AccessType: "ReadWriteOnce", | ||
| } | ||
|
|
||
| err := db.CreateOrUpdateAttachment(ctx, attachment) | ||
| assert.NoError(t, err) | ||
|
|
||
| err = db.DeleteAttachmentDisregardingAccessType(attachment.VolumeId, attachment.TargetPath, attachment.Node, attachment.BootID) | ||
| assert.NoError(t, err) | ||
|
|
||
| var result PvcAttachment | ||
| err = db.First(&result, "volume_id = ?", attachment.VolumeId).Error | ||
| assert.Error(t, err) | ||
| assert.Equal(t, gorm.ErrRecordNotFound, err) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| package db | ||
|
|
||
| import "context" | ||
|
|
||
| type Database interface { | ||
| GetAttachmentsByVolumeIdOrTargetPath(ctx context.Context, volumeId, targetPath *string) (*[]PvcAttachment, error) | ||
| CreateAttachment(ctx context.Context, attachment *PvcAttachment) error | ||
| UpdateAttachment(ctx context.Context, attachment *PvcAttachment) error | ||
| CreateOrUpdateAttachment(ctx context.Context, attachment *PvcAttachment) error | ||
| DeleteAttachment(ctx context.Context, attachment *PvcAttachment) error | ||
| DeleteAttachmentDisregardingAccessType(volumeId, targetPath, node, bootId string) error | ||
| } |