Refactor to introduce a storage interface#74
Open
ian-noaa wants to merge 5 commits into
Open
Conversation
This will let us add an S3 storage provider alongside the local storage provider.
This will keep memory use constant compared to io.ReadAll which reads everything into memory at once.
Collaborator
Author
|
@gopa-noaa when you have a minute, if you could review this one too, I'd appreciate it! |
There was a problem hiding this comment.
Pull request overview
This PR refactors stat-file ingestion to go through a new StorageProvider interface, enabling alternate backends (e.g., future S3) while keeping the existing local-file path flow working.
Changes:
- Introduces
pkg/storage.StorageProviderplus a local filesystem implementation (LocalProvider). - Refactors core processing to consume
(name, io.Reader)streams instead of opening files directly. - Adds/updates tests and reformats documentation.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| README.md | Markdown/doc formatting updates and config examples. |
| pkg/storage/provider.go | Adds StorageProvider abstraction for walking stat-file inputs. |
| pkg/storage/local.go | Implements local filesystem-backed provider. |
| pkg/storage/local_test.go | Adds unit tests for the local provider. |
| pkg/core/statToCbRun.go | Refactors processing entrypoints to accept a StorageProvider. |
| pkg/core/statFileToCbDocMetParser.go | Switches to streaming parse from io.Reader via bufio.Scanner. |
| pkg/core/statFileToCbDocMetParser_test.go | Adds a parser test and benchmark for the streaming parser. |
| pkg/core/procesdInput.go | Refactors processing to accept a StorageProvider (provider-based pipeline). |
| cmd/metjson2db/main.go | Wires CLI to use the local provider and provider-based core entrypoint. |
Comments suppressed due to low confidence (3)
pkg/storage/local_test.go:55
- Ignoring
os.WriteFileerrors can hide failures in test setup. Check and fail the test on write errors.
file1 := filepath.Join(dir, "a.stat")
_ = os.WriteFile(file1, []byte("content"), 0o644)
pkg/storage/local_test.go:76
- Ignoring
os.WriteFileerrors can hide failures in test setup. Check and fail the test on write errors.
file1 := filepath.Join(dir, "a.stat")
file2 := filepath.Join(dir, "b.stat")
_ = os.WriteFile(file1, []byte("data1"), 0o644)
_ = os.WriteFile(file2, []byte("data2"), 0o644)
pkg/storage/local_test.go:127
- Ignoring
os.WriteFileerrors can hide failures in test setup. Check and fail the test on write errors.
file1 := filepath.Join(dir, "a.stat")
file2 := filepath.Join(dir, "b.stat")
_ = os.WriteFile(file1, []byte("data1"), 0o644)
_ = os.WriteFile(file2, []byte("data2"), 0o644)
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+31
to
+40
| f, err := os.Open(path) | ||
| if err != nil { | ||
| return fmt.Errorf("opening %s: %w", path, err) | ||
| } | ||
| defer f.Close() | ||
|
|
||
| err = fn(path, f) | ||
| if err != nil { | ||
| return err | ||
| } |
Comment on lines
+38
to
+41
| if !scanner.Scan() { | ||
| return nil, fmt.Errorf("empty file or error reading header for %s: %w", name, scanner.Err()) | ||
| } | ||
| lines := strings.Split(string(rawData), "\n") | ||
| headerLine := lines[0] | ||
| headerLine := scanner.Text() |
Comment on lines
41
to
45
| state.AsyncWaitGroupFlushToDb.Add(1) | ||
| go func() { | ||
| defer state.AsyncWaitGroupFlushToDb.Done() | ||
| // conn := getDbConnection(credentials) | ||
| async.FlushToDbAsync(di) | ||
| }() |
Comment on lines
+164
to
171
| CREATE*JSON_DOC_ARCHIVE - METdadacb will create a gzip archive of Couchbase json documents from input stat files, | ||
| which can then be uploaded to Couchbase later using a cbimport command line tool. | ||
| Please note that the merge mode, when set using [overWriteData: false], is NOT available in CREATE_JSON_DOC_ARCHIVE mode, | ||
| since the cbimport tool will overwrite existing documents in the database that has same ID as incoming documents. | ||
| In this mode, the setting: | ||
| "jsonArchiveFilePathAndPrefix" :"/scratch/METjson2db_out_" | ||
| sets the folder and file prefix for the generated archive file. At the end of a succesfull run, the output file name | ||
| "jsonArchiveFilePathAndPrefix" :"/scratch/METjson2db_out*" | ||
| sets the folder and file prefix for the generated archive file. At the end of a succesfull run, the output file name | ||
| will looks like below (prefix + timestamp): |
Comment on lines
+8
to
+11
| // StorageProvider abstracts access to stat files from different storage backends. | ||
| type StorageProvider interface { | ||
| Walk(ctx context.Context, fn func(name string, r io.Reader) error) error | ||
| } |
Comment on lines
+15
to
+18
| file1 := filepath.Join(dir, "a.stat") | ||
| file2 := filepath.Join(dir, "b.stat") | ||
| _ = os.WriteFile(file1, []byte("header1\ndata1"), 0o644) | ||
| _ = os.WriteFile(file2, []byte("header2\ndata2"), 0o644) |
Comment on lines
+42
to
+58
| func BenchmarkParseStatFileContent(b *testing.B) { | ||
| filePath := "../../test_data/grid_stat_GFS_TMP_vs_ANLYS_TMP_P1000_anom_120000L_20240203_120000V.stat" | ||
| file, err := os.Open(filePath) | ||
| if err != nil { | ||
| b.Fatalf("Failed to open test file: %v", err) | ||
| } | ||
| defer file.Close() | ||
|
|
||
| b.ResetTimer() | ||
| b.ReportAllocs() // This is the magic flag! | ||
|
|
||
| for i := 0; i < b.N; i++ { | ||
| // Reset the file pointer to the beginning for each iteration | ||
| _, _ = file.Seek(0, 0) | ||
| _, _ = parseStatFileContent(filePath, file) | ||
| } | ||
| } |
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This will let us add an S3 storage provider alongside the local storage provider.