rory temp

staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java

@@ -134,7 +134,14 @@ public void deleteBatch(final TenantId tenantId, final BatchId batchId)
         }
 
         try {
+            LOGGER.info(
+                "Deleting completed batch with id {} for {}. Batch bytes: {}; Path: {}",
+                batchId,
+                tenantId,
+                calculateDirectorySize(batchPath),
+                batchPath);
             FileUtils.deleteDirectory(batchPath.toFile());
+            LOGGER.info("Deleted completed batch with id {} for {} at {}.", batchId, tenantId, batchPath);
         } catch (IOException ex) {
             throw new StagingException(ex);
         }
@@ -192,7 +199,14 @@ public List<String> saveFiles(final TenantId tenantId, final BatchId batchId, fi
                     LOGGER.debug("Reading loose file...");
                     try (final InputStream inStream = fileItemInput.getInputStream()) {
                         FileUtils.copyInputStreamToFile(inStream, targetFile);
-                        LOGGER.debug("Loose file written to {}", targetFile);
+                        LOGGER.info(
+                            "Staging service stored loose file part. Tenant: {}; Batch: {}; Field name: {}; "
+                                + "Stored file: {}; File bytes: {}",
+                            tenantId,
+                            batchId,
+                            filename,
+                            targetFile,
+                            Files.size(targetFile.toPath()));
                         fileNames.add(targetFileName);
                         binaryFilesUploaded.put(filename, targetFileName);
                     } catch (IOException ex) {
@@ -229,9 +243,13 @@ private void cleanupInProgressBatch(final File inProgressBatchFolderPath)
     private void completeInProgressBatch(final TenantId tenantId, final Path inProgressBatchFolderPath, final BatchId batchId)
         throws StagingException
     {
-        LOGGER.info("Completing batch with id {} for {}...", batchId, tenantId);
-
         final Path batchFolder = batchPathProvider.getPathForBatch(tenantId, batchId);
+        LOGGER.info(
+            "Completing batch with id {} for {}. Moving in-progress path {} to completed path {}.",
+            batchId,
+            tenantId,
+            inProgressBatchFolderPath,
+            batchFolder);
         if (batchFolder.toFile().exists()) {
             LOGGER.warn("Batch {} has been previously uploaded.  Removing previously uploaded batch...", batchId);
             try {
@@ -251,7 +269,41 @@ private void completeInProgressBatch(final TenantId tenantId, final Path inProgr
             throw new StagingException(ex);
         }
 
-        LOGGER.debug("Batch {} completed successfully.", batchId);
+        LOGGER.info(
+            "Batch {} completed successfully for {} at {}. Batch bytes: {}",
+            batchId,
+            tenantId,
+            batchFolder,
+            calculateDirectorySize(batchFolder));
+    }
+
+    private static long calculateDirectorySize(final Path directory)
+    {
+        try (final Stream<Path> paths = Files.walk(directory, FileVisitOption.FOLLOW_LINKS)) {
+            return paths
+                .filter(Files::isRegularFile)
+                .mapToLong(FileSystemDao::getFileSizeSafely)
+                .sum();
+        } catch (final IOException ex) {
+            LOGGER.info(
+                "Unable to calculate staged batch size. Directory: {}; Error: {}",
+                directory,
+                ex.getMessage());
+            return 0;
+        }
+    }
+
+    private static long getFileSizeSafely(final Path file)
+    {
+        try {
+            return Files.size(file);
+        } catch (final IOException ex) {
+            LOGGER.info(
+                "Unable to read staged file size while calculating batch size. File: {}; Error: {}",
+                file,
+                ex.getMessage());
+            return 0;
+        }
     }
 
     /**

staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/SubBatchWriter.java

@@ -46,6 +46,7 @@ public class SubBatchWriter implements AutoCloseable
     private final File inProgressBatchFolder;
     private final int subbatchSize;
     private OutputStream outStream;
+    private File currentSubBatch;
     //Track number of document files processed
     private int count = 0;
 
@@ -60,6 +61,7 @@ private void createSubBatchOutStream() throws StagingException
         //Make a new subbatch file
         final String subBatchFileName = BatchNameProvider.getSubBatchName();
         final File subBatch = inProgressBatchFolder.toPath().resolve(subBatchFileName).toFile();
+        currentSubBatch = subBatch;
         LOGGER.debug("Created new subbatchFile : {} ", subBatch);
         //Open new stream to start writing to subbatch file
 
@@ -91,10 +93,20 @@ public void writeDocumentFile(final InputStreamSupplier inputStreamSupplier,
             createSubBatchOutStream();
         }
 
+        final long subBatchBytesBeforeWrite = currentSubBatch.length();
         try (final InputStream inStream = inputStreamSupplier.get()) {
             try {
                 JsonMinifier.validateAndMinifyJson(inStream, outStream, storageRefFolderPath,
                                                    inprogressContentFolderPath, fieldValueSizeThreshold, binaryFilesUploaded);
+                outStream.flush();
+                final long subBatchBytesAfterWrite = currentSubBatch.length();
+                LOGGER.info(
+                    "Staging service wrote document JSON to subbatch. Document number in subbatch: {}; "
+                        + "Bytes written: {}; Subbatch bytes: {}; Subbatch file: {}",
+                    count + 1,
+                    subBatchBytesAfterWrite - subBatchBytesBeforeWrite,
+                    subBatchBytesAfterWrite,
+                    currentSubBatch);
                 count++;
             } catch (final JsonProcessingException | InvalidDocumentException ex) {
                 LOGGER.error("Error staging document", ex);
@@ -120,6 +132,7 @@ public void close() throws Exception
             outStream.close();
         }
         outStream = null;
+        currentSubBatch = null;
     }
 
 }

staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java

@@ -28,6 +28,8 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.InvalidPathException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Map;
@@ -110,6 +112,7 @@ private static void processJsonTokens(final JsonParser parser,
         boolean updateReference = false;
         // mark if the updateReference is for a file that has been uploaded
         boolean isLocalRefFile = false;
+        String currentDocumentFieldName = null;
         while ((token = parser.nextToken()) != null) {
             switch (token) {
                 case FIELD_NAME:
@@ -126,6 +129,7 @@ private static void processJsonTokens(final JsonParser parser,
                         bufferData = false;
                         bufferEncoding = false;
                         pauseWriting = false;
+                        currentDocumentFieldName = parser.getText();
                     }
                     break;
                 case VALUE_FALSE:
@@ -157,7 +161,13 @@ private static void processJsonTokens(final JsonParser parser,
                         if (dataBuffer != null && dataBuffer.getBytes().length > fieldValueSizeThreshold) {
                             // write it out to a loose file
                             final String fileName = RandomStringUtils.randomAlphanumeric(10);
-                            final String contentFileName = writeDataToFile(dataBuffer, fileName, inprogressContentFolderPath, encodingBuffer);
+                            final String contentFileName = writeDataToFile(
+                                dataBuffer,
+                                fileName,
+                                inprogressContentFolderPath,
+                                encodingBuffer,
+                                currentDocumentFieldName,
+                                fieldValueSizeThreshold);
                             dataBuffer = contentFileName;
                             encodingBuffer = STORAGE_REF;
                             updateReference = true;
@@ -175,6 +185,8 @@ private static void processJsonTokens(final JsonParser parser,
                             } else {
                                 dataBuffer = storageRefPath + "/" + dataBuffer;
                             }
+                        } else if (STORAGE_REF.equalsIgnoreCase(encodingBuffer)) {
+                            logExistingStorageReference(dataBuffer, currentDocumentFieldName);
                         }
                         gen.writeFieldName(DATA_FIELD);
                         gen.writeString(dataBuffer);
@@ -205,21 +217,83 @@ private static void processJsonTokens(final JsonParser parser,
         }
     }
 
-    private static String writeDataToFile(final String data, final String fileName,
-                                          final String inprogressContentFolderPath, final String encoding) throws IOException
+    private static String writeDataToFile(final String data,
+                                          final String fileName,
+                                          final String inprogressContentFolderPath,
+                                          final String encoding,
+                                          final String documentFieldName,
+                                          final int fieldValueSizeThreshold) throws IOException
     {
         String contentFileName = fileName;
+        Path targetFile = null;
+        long inputBytes = data.getBytes(StandardCharsets.UTF_8).length;
         if (encoding == null || encoding.equalsIgnoreCase(UTF8_ENCODING)) {
             contentFileName = fileName + TXT_ENTENSION;
-            final Path targetFile = Paths.get(inprogressContentFolderPath, contentFileName);
+            targetFile = Paths.get(inprogressContentFolderPath, contentFileName);
             FileUtils.writeStringToFile(targetFile.toFile(), data, StandardCharsets.UTF_8);
         } else if (encoding.equalsIgnoreCase(BASE64_ENCODING)) {
             contentFileName = fileName + BINARY_ENTENSION;
-            final Path targetFile = Paths.get(inprogressContentFolderPath, contentFileName);
+            targetFile = Paths.get(inprogressContentFolderPath, contentFileName);
             // If encoding is base64, write file after base64 decoding the data
             final byte[] decodedData = Base64.decodeBase64(data);
+            inputBytes = decodedData.length;
             FileUtils.writeByteArrayToFile(targetFile.toFile(), decodedData);
         }
+        if (targetFile != null) {
+            LOGGER.info(
+                "Staging service externalized JSON field data to storage_ref. Field: {}; Encoding: {}; "
+                    + "Input bytes: {}; File bytes: {}; Threshold bytes: {}; File: {}",
+                documentFieldName,
+                encoding == null ? UTF8_ENCODING : encoding,
+                inputBytes,
+                Files.size(targetFile),
+                fieldValueSizeThreshold,
+                targetFile);
+        }
         return contentFileName;
     }
+
+    private static void logExistingStorageReference(final String storageReference, final String documentFieldName)
+    {
+        if (storageReference == null) {
+            return;
+        }
+
+        final Path storageReferencePath;
+        try {
+            storageReferencePath = Paths.get(storageReference);
+        } catch (final InvalidPathException ex) {
+            LOGGER.info(
+                "Staging service observed existing storage_ref JSON field data but the reference is not a local path. "
+                    + "Field: {}; Reference: {}; Error: {}",
+                documentFieldName,
+                storageReference,
+                ex.getMessage());
+            return;
+        }
+
+        if (Files.exists(storageReferencePath)) {
+            try {
+                LOGGER.info(
+                    "Staging service observed existing storage_ref JSON field data. "
+                        + "Field: {}; File bytes: {}; File: {}",
+                    documentFieldName,
+                    Files.size(storageReferencePath),
+                    storageReferencePath);
+            } catch (final IOException ex) {
+                LOGGER.info(
+                    "Staging service observed existing storage_ref JSON field data but could not read file size. "
+                        + "Field: {}; File: {}; Error: {}",
+                    documentFieldName,
+                    storageReferencePath,
+                    ex.getMessage());
+            }
+        } else {
+            LOGGER.info(
+                "Staging service observed existing storage_ref JSON field data but the file was not present. "
+                    + "Field: {}; File: {}",
+                documentFieldName,
+                storageReferencePath);
+        }
+    }
 }