Skip to content

Commit 05b39a4

Browse files
saratchebrolusaratchebrolu
authored
Update the photo and video importers of Koofr and Backblaze to log bytes (#1195)
* Update the photo and video importers of Koofr and Backblaze to log bytes transferred Summary: Updated the photo and video importers of Koofr and Backblaze to log the number of bytes transferred * Updated * Fix formatting
1 parent bde428a commit 05b39a4

File tree

8 files changed

+213
-151
lines changed

8 files changed

+213
-151
lines changed

‎extensions/data-transfer/portability-data-transfer-backblaze/src/main/java/org/datatransferproject/datatransfer/backblaze/photos/BackblazePhotosImporter.java‎

Lines changed: 18 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
import java.io.IOException;
2323
import java.io.InputStream;
2424
import java.util.UUID;
25+
import java.util.concurrent.atomic.LongAdder;
26+
2527
import org.datatransferproject.api.launcher.Monitor;
2628
import org.datatransferproject.datatransfer.backblaze.common.BackblazeDataTransferClient;
2729
import org.datatransferproject.datatransfer.backblaze.common.BackblazeDataTransferClientFactory;
@@ -80,14 +82,23 @@ public ImportResult importItem(
8082
}
8183
}
8284

85+
final LongAdder totalImportedFilesSizes = new LongAdder();
8386
if (data.getPhotos() != null && data.getPhotos().size() > 0) {
8487
for (PhotoModel photo : data.getPhotos()) {
8588
idempotentExecutor.importAndSwallowIOExceptions(
86-
photo, p -> importSinglePhoto(idempotentExecutor, b2Client, jobId, p));
89+
photo,
90+
p -> {
91+
ItemImportResult<String> fileImportResult =
92+
importSinglePhoto(idempotentExecutor, b2Client, jobId, p);
93+
if (fileImportResult.hasBytes()) {
94+
totalImportedFilesSizes.add(fileImportResult.getBytes());
95+
}
96+
return fileImportResult;
97+
});
8798
}
8899
}
89100

90-
return ImportResult.OK;
101+
return ImportResult.OK.copyWithBytes(totalImportedFilesSizes.longValue());
91102
}
92103

93104
private ItemImportResult<String> importSinglePhoto(
@@ -115,8 +126,11 @@ private ItemImportResult<String> importSinglePhoto(
115126
} catch (Exception e) {
116127
// Swallow the exception caused by Remove data so that existing flows continue
117128
monitor.info(
118-
() -> format("Exception swallowed while removing data for jobId %s, localPath %s",
119-
jobId, photo.getFetchableUrl()), e);
129+
() ->
130+
format(
131+
"Exception swallowed while removing data for jobId %s, localPath %s",
132+
jobId, photo.getFetchableUrl()),
133+
e);
120134
}
121135

122136
return ItemImportResult.success(response, size);

‎extensions/data-transfer/portability-data-transfer-backblaze/src/main/java/org/datatransferproject/datatransfer/backblaze/videos/BackblazeVideosImporter.java‎

Lines changed: 16 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121
import java.io.IOException;
2222
import java.io.InputStream;
2323
import java.util.UUID;
24+
import java.util.concurrent.atomic.LongAdder;
25+
2426
import org.datatransferproject.api.launcher.Monitor;
2527
import org.datatransferproject.datatransfer.backblaze.common.BackblazeDataTransferClient;
2628
import org.datatransferproject.datatransfer.backblaze.common.BackblazeDataTransferClientFactory;
@@ -69,29 +71,32 @@ public ImportResult importItem(
6971

7072
BackblazeDataTransferClient b2Client = b2ClientFactory.getOrCreateB2Client(jobId, authData);
7173

74+
final LongAdder totalImportedFilesSizes = new LongAdder();
7275
if (data.getVideos() != null && data.getVideos().size() > 0) {
7376
for (VideoModel video : data.getVideos()) {
7477
idempotentExecutor.importAndSwallowIOExceptions(
7578
video,
76-
v -> importSingleVideo(jobId, b2Client, v));
79+
v -> {
80+
ItemImportResult<String> fileImportResult = importSingleVideo(jobId, b2Client, v);
81+
if (fileImportResult.hasBytes()) {
82+
totalImportedFilesSizes.add(fileImportResult.getBytes());
83+
}
84+
return fileImportResult;
85+
});
7786
}
7887
}
7988

80-
return ImportResult.OK;
89+
return ImportResult.OK.copyWithBytes(totalImportedFilesSizes.longValue());
8190
}
8291

8392
private ItemImportResult<String> importSingleVideo(
84-
UUID jobId,
85-
BackblazeDataTransferClient b2Client,
86-
VideoModel video)
87-
throws IOException {
93+
UUID jobId, BackblazeDataTransferClient b2Client, VideoModel video) throws IOException {
8894
try (InputStream videoFileStream =
8995
connectionProvider.getInputStreamForItem(jobId, video).getStream()) {
90-
File file = jobStore
91-
.getTempFileFromInputStream(videoFileStream, video.getDataId(), ".mp4");
92-
String res = b2Client.uploadFile(
93-
String.format("%s/%s.mp4", VIDEO_TRANSFER_MAIN_FOLDER, video.getDataId()),
94-
file);
96+
File file = jobStore.getTempFileFromInputStream(videoFileStream, video.getDataId(), ".mp4");
97+
String res =
98+
b2Client.uploadFile(
99+
String.format("%s/%s.mp4", VIDEO_TRANSFER_MAIN_FOLDER, video.getDataId()), file);
95100
return ItemImportResult.success(res, file.length());
96101
} catch (FileNotFoundException e) {
97102
monitor.info(

‎extensions/data-transfer/portability-data-transfer-backblaze/src/test/java/org/datatransferproject/datatransfer/backblaze/photos/BackblazePhotosImporterTest.java‎

Lines changed: 14 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -49,13 +49,13 @@
4949
import org.mockito.ArgumentCaptor;
5050

5151
public class BackblazePhotosImporterTest {
52-
Monitor monitor;
53-
TemporaryPerJobDataStore dataStore;
54-
ConnectionProvider streamProvider;
55-
BackblazeDataTransferClientFactory clientFactory;
56-
IdempotentImportExecutor executor;
57-
TokenSecretAuthData authData;
58-
BackblazeDataTransferClient client;
52+
Monitor monitor;
53+
TemporaryPerJobDataStore dataStore;
54+
ConnectionProvider streamProvider;
55+
BackblazeDataTransferClientFactory clientFactory;
56+
IdempotentImportExecutor executor;
57+
TokenSecretAuthData authData;
58+
BackblazeDataTransferClient client;
5959

6060
@BeforeEach
6161
public void setUp() {
@@ -68,8 +68,7 @@ public void setUp() {
6868
client = mock(BackblazeDataTransferClient.class);
6969
}
7070

71-
@TempDir
72-
public Path folder;
71+
@TempDir public Path folder;
7372

7473
@Test
7574
public void testNullData() throws Exception {
@@ -88,7 +87,7 @@ public void testNullPhotosAndAlbums() throws Exception {
8887
BackblazePhotosImporter sut =
8988
new BackblazePhotosImporter(monitor, dataStore, streamProvider, clientFactory);
9089
ImportResult result = sut.importItem(UUID.randomUUID(), executor, authData, data);
91-
assertEquals(ImportResult.OK, result);
90+
assertEquals(ImportResult.ResultType.OK, result.getType());
9291
}
9392

9493
@Test
@@ -100,7 +99,7 @@ public void testEmptyPhotosAndAlbums() throws Exception {
10099
BackblazePhotosImporter sut =
101100
new BackblazePhotosImporter(monitor, dataStore, streamProvider, clientFactory);
102101
ImportResult result = sut.importItem(UUID.randomUUID(), executor, authData, data);
103-
assertEquals(ImportResult.OK, result);
102+
assertEquals(ImportResult.ResultType.OK, result.getType());
104103
}
105104

106105
@Test
@@ -113,8 +112,8 @@ public void testImportPhoto() throws Exception {
113112
String response = "response";
114113
UUID jobId = UUID.randomUUID();
115114
PhotoModel photoModel = new PhotoModel(title, photoUrl, "", "", dataId, albumId, false);
116-
PhotosContainerResource data = new PhotosContainerResource(Collections.emptyList(),
117-
Collections.singletonList(photoModel));
115+
PhotosContainerResource data =
116+
new PhotosContainerResource(Collections.emptyList(), Collections.singletonList(photoModel));
118117

119118
when(executor.getCachedValue(albumId)).thenReturn(albumName);
120119

@@ -131,8 +130,8 @@ public void testImportPhoto() throws Exception {
131130
new BackblazePhotosImporter(monitor, dataStore, streamProvider, clientFactory);
132131
sut.importItem(jobId, executor, authData, data);
133132

134-
ArgumentCaptor<ImportFunction<PhotoModel, String>> importCapture = ArgumentCaptor.forClass(
135-
ImportFunction.class);
133+
ArgumentCaptor<ImportFunction<PhotoModel, String>> importCapture =
134+
ArgumentCaptor.forClass(ImportFunction.class);
136135
verify(executor, times(1))
137136
.importAndSwallowIOExceptions(eq(photoModel), importCapture.capture());
138137

‎extensions/data-transfer/portability-data-transfer-backblaze/src/test/java/org/datatransferproject/datatransfer/backblaze/videos/BackblazeVideosImporterTest.java‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -54,8 +54,7 @@ public class BackblazeVideosImporterTest {
5454
IdempotentImportExecutor executor;
5555
TokenSecretAuthData authData;
5656
BackblazeDataTransferClient client;
57-
@TempDir
58-
public Path folder;
57+
@TempDir public Path folder;
5958

6059
@BeforeEach
6160
public void setUp() {
@@ -84,7 +83,7 @@ public void testNullVideos() throws Exception {
8483
BackblazeVideosImporter sut =
8584
new BackblazeVideosImporter(monitor, dataStore, streamProvider, clientFactory);
8685
ImportResult result = sut.importItem(UUID.randomUUID(), executor, authData, data);
87-
assertEquals(ImportResult.OK, result);
86+
assertEquals(ImportResult.ResultType.OK, result.getType());
8887
}
8988

9089
@Test
@@ -95,7 +94,7 @@ public void testEmptyVideos() throws Exception {
9594
BackblazeVideosImporter sut =
9695
new BackblazeVideosImporter(monitor, dataStore, streamProvider, clientFactory);
9796
ImportResult result = sut.importItem(UUID.randomUUID(), executor, authData, data);
98-
assertEquals(ImportResult.OK, result);
97+
assertEquals(ImportResult.ResultType.OK, result.getType());
9998
}
10099

101100
@Test
@@ -130,8 +129,8 @@ public void testImportVideo() throws Exception {
130129
new BackblazeVideosImporter(monitor, dataStore, streamProvider, clientFactory);
131130
sut.importItem(jobId, executor, authData, data);
132131

133-
ArgumentCaptor<ImportFunction<VideoModel, String>> importCapture = ArgumentCaptor.forClass(
134-
ImportFunction.class);
132+
ArgumentCaptor<ImportFunction<VideoModel, String>> importCapture =
133+
ArgumentCaptor.forClass(ImportFunction.class);
135134
verify(executor, times(1))
136135
.importAndSwallowIOExceptions(eq(videoObject), importCapture.capture());
137136

‎extensions/data-transfer/portability-data-transfer-koofr/src/main/java/org/datatransferproject/transfer/koofr/photos/KoofrPhotosImporter.java‎

Lines changed: 80 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -15,16 +15,16 @@
1515
*/
1616
package org.datatransferproject.transfer.koofr.photos;
1717

18-
import java.io.BufferedInputStream;
1918
import java.io.ByteArrayInputStream;
2019
import java.io.IOException;
2120
import java.io.InputStream;
22-
import java.net.HttpURLConnection;
2321
import java.text.SimpleDateFormat;
2422
import java.util.Date;
2523
import java.util.HashMap;
2624
import java.util.TimeZone;
2725
import java.util.UUID;
26+
import java.util.concurrent.atomic.LongAdder;
27+
2828
import org.apache.commons.imaging.Imaging;
2929
import org.apache.commons.imaging.common.ImageMetadata;
3030
import org.apache.commons.imaging.formats.jpeg.JpegImageMetadata;
@@ -34,7 +34,9 @@
3434
import org.datatransferproject.api.launcher.Monitor;
3535
import org.datatransferproject.spi.cloud.connection.ConnectionProvider;
3636
import org.datatransferproject.spi.cloud.storage.JobStore;
37+
import org.datatransferproject.spi.cloud.storage.TemporaryPerJobDataStore.InputStreamWrapper;
3738
import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutor;
39+
import org.datatransferproject.spi.transfer.idempotentexecutor.ItemImportResult;
3840
import org.datatransferproject.spi.transfer.provider.ImportResult;
3941
import org.datatransferproject.spi.transfer.provider.Importer;
4042
import org.datatransferproject.spi.transfer.types.DestinationMemoryFullException;
@@ -103,13 +105,20 @@ public ImportResult importItem(
103105
album.getId(), album.getName(), () -> createAlbumFolder(album, koofrClient));
104106
}
105107

108+
final LongAdder totalImportedFilesSizes = new LongAdder();
106109
for (PhotoModel photoModel : resource.getPhotos()) {
107-
idempotentImportExecutor.executeAndSwallowIOExceptions(
108-
photoModel.getIdempotentId(),
109-
photoModel.getTitle(),
110-
() -> importSinglePhoto(photoModel, jobId, idempotentImportExecutor, koofrClient));
110+
idempotentImportExecutor.importAndSwallowIOExceptions(
111+
photoModel,
112+
photo -> {
113+
ItemImportResult<String> fileImportResult =
114+
importSinglePhoto(photoModel, jobId, idempotentImportExecutor, koofrClient);
115+
if (fileImportResult != null && fileImportResult.hasBytes()) {
116+
totalImportedFilesSizes.add(fileImportResult.getBytes());
117+
}
118+
return fileImportResult;
119+
});
111120
}
112-
return ImportResult.OK;
121+
return ImportResult.OK.copyWithBytes(totalImportedFilesSizes.longValue());
113122
}
114123

115124
private String createAlbumFolder(PhotoAlbum album, KoofrClient koofrClient)
@@ -132,61 +141,89 @@ private String createAlbumFolder(PhotoAlbum album, KoofrClient koofrClient)
132141
return fullPath;
133142
}
134143

135-
private String importSinglePhoto(
144+
private ItemImportResult<String> importSinglePhoto(
136145
PhotoModel photo,
137146
UUID jobId,
138147
IdempotentImportExecutor idempotentImportExecutor,
139148
KoofrClient koofrClient)
140149
throws IOException, InvalidTokenException, DestinationMemoryFullException {
141150
monitor.debug(() -> String.format("Import single photo %s", photo.getTitle()));
151+
Long size = null;
152+
try {
153+
InputStreamWrapper inputStreamWrapper =
154+
connectionProvider.getInputStreamForItem(jobId, photo);
155+
ItemImportResult<String> response;
142156

143-
try (InputStream inputStream =
144-
connectionProvider.getInputStreamForItem(jobId, photo).getStream()) {
145-
final byte[] bytes = IOUtils.toByteArray(inputStream);
146-
147-
Date dateCreated = getDateCreated(photo, bytes);
148-
149-
String title = buildPhotoTitle(jobId, photo.getTitle(), dateCreated);
150-
String description = KoofrClient.trimDescription(photo.getDescription());
151-
152-
String parentPath = idempotentImportExecutor.getCachedValue(photo.getAlbumId());
153-
String fullPath = parentPath + "/" + title;
157+
try (InputStream inputStream = inputStreamWrapper.getStream()) {
158+
final byte[] bytes = IOUtils.toByteArray(inputStream);
154159

155-
if (koofrClient.fileExists(fullPath)) {
156-
monitor.debug(() -> String.format("Photo already exists %s", photo.getTitle()));
160+
Date dateCreated = getDateCreated(photo, bytes);
157161

158-
return fullPath;
159-
}
162+
String title = buildPhotoTitle(jobId, photo.getTitle(), dateCreated);
163+
String description = KoofrClient.trimDescription(photo.getDescription());
160164

161-
final ByteArrayInputStream inMemoryInputStream = new ByteArrayInputStream(bytes);
165+
String parentPath = idempotentImportExecutor.getCachedValue(photo.getAlbumId());
166+
String fullPath = parentPath + "/" + title;
162167

163-
String response;
168+
if (koofrClient.fileExists(fullPath)) {
169+
monitor.debug(() -> String.format("Photo already exists %s", photo.getTitle()));
164170

165-
try {
166-
response = koofrClient.uploadFile(
167-
parentPath, title, inMemoryInputStream, photo.getMediaType(), dateCreated, description);
168-
} catch (KoofrClientIOException e) {
169-
if (e.getCode() == 404) {
170-
monitor.info(() -> String.format("Can't find album during importSingleItem for id: %s", photo.getDataId()), e);
171-
response = "skipped-"+photo.getDataId();
171+
return ItemImportResult.success(fullPath);
172172
}
173-
else {
174-
throw e;
173+
174+
final ByteArrayInputStream inMemoryInputStream = new ByteArrayInputStream(bytes);
175+
176+
try {
177+
long inputStreamBytes = inputStreamWrapper.getBytes();
178+
String stringResult =
179+
koofrClient.uploadFile(
180+
parentPath,
181+
title,
182+
inMemoryInputStream,
183+
photo.getMediaType(),
184+
dateCreated,
185+
description);
186+
if (stringResult != null && !stringResult.isEmpty()) {
187+
response = ItemImportResult.success(stringResult, inputStreamBytes);
188+
} else {
189+
response =
190+
ItemImportResult.success(
191+
String.format(SKIPPED_FILE_RESULT_FORMAT, photo.getDataId()));
192+
}
193+
size = inputStreamBytes;
194+
} catch (KoofrClientIOException exception) {
195+
if (exception.getCode() == 404) {
196+
monitor.info(
197+
() ->
198+
String.format(
199+
"Can't find album during importSingleItem for id: %s", photo.getDataId()),
200+
exception);
201+
response =
202+
ItemImportResult.success(
203+
String.format(SKIPPED_FILE_RESULT_FORMAT, photo.getDataId()));
204+
} else {
205+
return ItemImportResult.error(exception, size);
206+
}
175207
}
176-
}
177208

178-
try {
179-
if (photo.isInTempStore()) {
180-
jobStore.removeData(jobId, photo.getFetchableUrl());
209+
try {
210+
if (photo.isInTempStore()) {
211+
jobStore.removeData(jobId, photo.getFetchableUrl());
212+
}
213+
} catch (Exception e) {
214+
// Swallow the exception caused by Remove data so that existing flows continue
215+
monitor.info(
216+
() ->
217+
format(
218+
"Exception swallowed while removing data for jobId %s, localPath %s",
219+
jobId, photo.getFetchableUrl()),
220+
e);
181221
}
182-
} catch (Exception e) {
183-
// Swallow the exception caused by Remove data so that existing flows continue
184-
monitor.info(
185-
() -> format("Exception swallowed while removing data for jobId %s, localPath %s",
186-
jobId, photo.getFetchableUrl()), e);
187222
}
188223

189224
return response;
225+
} catch (KoofrClientIOException exception) {
226+
return ItemImportResult.error(exception, size);
190227
}
191228
}
192229

0 commit comments

Comments
 (0)