diff --git a/router/batchrouter/asyncdestinationmanager/bing-ads/audience/types.go b/router/batchrouter/asyncdestinationmanager/bing-ads/audience/types.go index ddeb9cf77f..b7057a0a5c 100644 --- a/router/batchrouter/asyncdestinationmanager/bing-ads/audience/types.go +++ b/router/batchrouter/asyncdestinationmanager/bing-ads/audience/types.go @@ -51,6 +51,7 @@ type DestinationConfig struct { type ActionFileInfo struct { Action string CSVWriter *csv.Writer + CSVFile *os.File CSVFilePath string ZipFilePath string SuccessfulJobIDs []int64 diff --git a/router/batchrouter/asyncdestinationmanager/bing-ads/audience/util.go b/router/batchrouter/asyncdestinationmanager/bing-ads/audience/util.go index b59fbe082a..ae1eb60681 100644 --- a/router/batchrouter/asyncdestinationmanager/bing-ads/audience/util.go +++ b/router/batchrouter/asyncdestinationmanager/bing-ads/audience/util.go @@ -50,20 +50,27 @@ func createActionFile(audienceId, actionType string) (*ActionFileInfo, error) { } csvWriter, err := CreateActionFileTemplate(csvFile, audienceId, actionType) if err != nil { + _ = csvFile.Close() + _ = os.Remove(csvFilePath) return nil, err } return &ActionFileInfo{ Action: actionType, ZipFilePath: zipFilePath, CSVFilePath: csvFilePath, + CSVFile: csvFile, CSVWriter: csvWriter, }, nil } func convertCsvToZip(actionFile *ActionFileInfo) error { + if actionFile.CSVFile != nil { + _ = actionFile.CSVFile.Close() + actionFile.CSVFile = nil + } if actionFile.EventCount == 0 { - os.Remove(actionFile.CSVFilePath) - os.Remove(actionFile.ZipFilePath) + _ = os.Remove(actionFile.CSVFilePath) + _ = os.Remove(actionFile.ZipFilePath) return nil } zipFile, err := os.Create(actionFile.ZipFilePath) @@ -82,6 +89,8 @@ func convertCsvToZip(actionFile *ActionFileInfo) error { if err != nil { return err } + defer csvFile.Close() + if _, err := csvFile.Seek(0, 0); err != nil { return err } @@ -143,9 +152,18 @@ func (b *BingAdsBulkUploader) createZipFile(filePath, audienceId string) ([]*Act defer textFile.Close() actionFiles := map[string]*ActionFileInfo{} + closeActionFiles := func() { + for _, af := range actionFiles { + if af != nil && af.CSVFile != nil { + _ = af.CSVFile.Close() + af.CSVFile = nil + } + } + } for _, actionType := range actionTypes { actionFiles[actionType], err = createActionFile(audienceId, actionType) if err != nil { + closeActionFiles() return nil, err } } @@ -155,6 +173,7 @@ func (b *BingAdsBulkUploader) createZipFile(filePath, audienceId string) ([]*Act line := scanner.Text() var data Data if err := jsonrs.Unmarshal([]byte(line), &data); err != nil { + closeActionFiles() return nil, err } @@ -167,12 +186,14 @@ func (b *BingAdsBulkUploader) createZipFile(filePath, audienceId string) ([]*Act actionFile := actionFiles[data.Message.Action] err := b.populateZipFile(actionFile, audienceId, line, data) if err != nil { + closeActionFiles() return nil, err } } scannerErr := scanner.Err() if scannerErr != nil { + closeActionFiles() return nil, scannerErr } actionFilesList := []*ActionFileInfo{} diff --git a/router/batchrouter/asyncdestinationmanager/bing-ads/audience/util_close_test.go b/router/batchrouter/asyncdestinationmanager/bing-ads/audience/util_close_test.go new file mode 100644 index 0000000000..1ee36860e9 --- /dev/null +++ b/router/batchrouter/asyncdestinationmanager/bing-ads/audience/util_close_test.go @@ -0,0 +1,60 @@ +package audience + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/rudderlabs/rudder-server/utils/misc" +) + +func ensureAsyncDestLogsDir(t *testing.T) { + t.Helper() + tmpDirPath, err := misc.GetTmpDir() + require.NoError(t, err) + require.NoError(t, os.MkdirAll(filepath.Join(tmpDirPath, misc.RudderAsyncDestinationLogs), 0o755)) +} + +func TestConvertCsvToZipClosesHandles(t *testing.T) { + t.Parallel() + ensureAsyncDestLogsDir(t) + + t.Run("empty event count closes write handle and removes csv", func(t *testing.T) { + t.Parallel() + + actionFile, err := createActionFile("aud-1", "Add") + require.NoError(t, err) + require.NotNil(t, actionFile.CSVFile) + + writeHandle := actionFile.CSVFile + require.NoError(t, convertCsvToZip(actionFile)) + require.Nil(t, actionFile.CSVFile) + + _, writeErr := writeHandle.Write([]byte("x")) + require.Error(t, writeErr) + _, statErr := os.Stat(actionFile.CSVFilePath) + require.True(t, os.IsNotExist(statErr)) + }) + + t.Run("non-empty event count closes write handle and creates zip", func(t *testing.T) { + t.Parallel() + + actionFile, err := createActionFile("aud-1", "Add") + require.NoError(t, err) + actionFile.EventCount = 1 + actionFile.CSVWriter.Flush() + + writeHandle := actionFile.CSVFile + require.NoError(t, convertCsvToZip(actionFile)) + require.Nil(t, actionFile.CSVFile) + + _, writeErr := writeHandle.Write([]byte("x")) + require.Error(t, writeErr) + + _, err = os.Stat(actionFile.ZipFilePath) + require.NoError(t, err) + require.NoError(t, os.Remove(actionFile.ZipFilePath)) + }) +} diff --git a/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/types.go b/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/types.go index dec2b2399d..b9c327673e 100644 --- a/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/types.go +++ b/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/types.go @@ -3,6 +3,7 @@ package offline_conversions import ( "encoding/csv" "encoding/json" + "os" "github.com/rudderlabs/bing-ads-go-sdk/bingads" "github.com/rudderlabs/rudder-go-kit/logger" @@ -42,6 +43,7 @@ type DestinationConfig struct { type ActionFileInfo struct { Action string CSVWriter *csv.Writer + CSVFile *os.File CSVFilePath string ZipFilePath string SuccessfulJobIDs []int64 diff --git a/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util.go b/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util.go index a85290c4af..e3d3adfb99 100644 --- a/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util.go +++ b/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util.go @@ -73,20 +73,27 @@ func createActionFile(actionType string) (*ActionFileInfo, error) { } csvWriter, err := CreateActionFileTemplate(csvFile, actionType) if err != nil { + _ = csvFile.Close() + _ = os.Remove(csvFilePath) return nil, err } return &ActionFileInfo{ Action: actionType, ZipFilePath: zipFilePath, CSVFilePath: csvFilePath, + CSVFile: csvFile, CSVWriter: csvWriter, }, nil } func convertCsvToZip(actionFile *ActionFileInfo) error { + if actionFile.CSVFile != nil { + _ = actionFile.CSVFile.Close() + actionFile.CSVFile = nil + } if actionFile.EventCount == 0 { - os.Remove(actionFile.CSVFilePath) - os.Remove(actionFile.ZipFilePath) + _ = os.Remove(actionFile.CSVFilePath) + _ = os.Remove(actionFile.ZipFilePath) return nil } zipFile, err := os.Create(actionFile.ZipFilePath) @@ -105,6 +112,8 @@ func convertCsvToZip(actionFile *ActionFileInfo) error { if err != nil { return err } + defer csvFile.Close() + if _, err := csvFile.Seek(0, 0); err != nil { return err } @@ -176,9 +185,18 @@ func (b *BingAdsBulkUploader) createZipFile(filePath string) ([]*ActionFileInfo, } defer textFile.Close() actionFiles := map[string]*ActionFileInfo{} + closeActionFiles := func() { + for _, af := range actionFiles { + if af != nil && af.CSVFile != nil { + _ = af.CSVFile.Close() + af.CSVFile = nil + } + } + } for _, actionType := range actionTypes { actionFiles[actionType], err = createActionFile(actionType) if err != nil { + closeActionFiles() return nil, err } } @@ -188,17 +206,20 @@ func (b *BingAdsBulkUploader) createZipFile(filePath string) ([]*ActionFileInfo, line := scanner.Text() var data Data if err := jsonrs.Unmarshal([]byte(line), &data); err != nil { + closeActionFiles() return nil, err } actionFile := actionFiles[data.Message.Action] err := b.populateZipFile(actionFile, line, data) if err != nil { + closeActionFiles() return nil, err } } scannerErr := scanner.Err() if scannerErr != nil { + closeActionFiles() return nil, scannerErr } actionFilesList := []*ActionFileInfo{} diff --git a/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util_close_test.go b/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util_close_test.go new file mode 100644 index 0000000000..c72c6033c6 --- /dev/null +++ b/router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util_close_test.go @@ -0,0 +1,60 @@ +package offline_conversions + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/rudderlabs/rudder-server/utils/misc" +) + +func ensureAsyncDestLogsDir(t *testing.T) { + t.Helper() + tmpDirPath, err := misc.GetTmpDir() + require.NoError(t, err) + require.NoError(t, os.MkdirAll(filepath.Join(tmpDirPath, misc.RudderAsyncDestinationLogs), 0o755)) +} + +func TestConvertCsvToZipClosesHandles(t *testing.T) { + t.Parallel() + ensureAsyncDestLogsDir(t) + + t.Run("empty event count closes write handle and removes csv", func(t *testing.T) { + t.Parallel() + + actionFile, err := createActionFile("insert") + require.NoError(t, err) + require.NotNil(t, actionFile.CSVFile) + + writeHandle := actionFile.CSVFile + require.NoError(t, convertCsvToZip(actionFile)) + require.Nil(t, actionFile.CSVFile) + + _, writeErr := writeHandle.Write([]byte("x")) + require.Error(t, writeErr) + _, statErr := os.Stat(actionFile.CSVFilePath) + require.True(t, os.IsNotExist(statErr)) + }) + + t.Run("non-empty event count closes write handle and creates zip", func(t *testing.T) { + t.Parallel() + + actionFile, err := createActionFile("insert") + require.NoError(t, err) + actionFile.EventCount = 1 + actionFile.CSVWriter.Flush() + + writeHandle := actionFile.CSVFile + require.NoError(t, convertCsvToZip(actionFile)) + require.Nil(t, actionFile.CSVFile) + + _, writeErr := writeHandle.Write([]byte("x")) + require.Error(t, writeErr) + + _, err = os.Stat(actionFile.ZipFilePath) + require.NoError(t, err) + require.NoError(t, os.Remove(actionFile.ZipFilePath)) + }) +}