fix: free up cache space when files get deleted (#198)

Clear processing file information when files get deleted

src/integrationtests/java/com/aws/greengrass/integrationtests/logmanager/LogManagerTest.java

@@ -5,6 +5,7 @@
 
 package com.aws.greengrass.integrationtests.logmanager;
 
+import com.aws.greengrass.config.Topics;
 import com.aws.greengrass.dependency.State;
 import com.aws.greengrass.deployment.DeviceConfiguration;
 import com.aws.greengrass.deployment.exceptions.DeviceConfigurationException;
@@ -19,12 +20,14 @@
 import com.aws.greengrass.logmanager.model.ComponentLogConfiguration;
 import com.aws.greengrass.logmanager.model.EventType;
 import com.aws.greengrass.logmanager.model.LogFileGroup;
+import com.aws.greengrass.logmanager.model.ProcessingFiles;
 import com.aws.greengrass.testcommons.testutilities.GGExtension;
 import com.aws.greengrass.util.exceptions.TLSAuthException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.extension.ExtensionContext;
@@ -47,6 +50,7 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.text.SimpleDateFormat;
+import java.time.Duration;
 import java.time.Instant;
 import java.time.format.DateTimeParseException;
 import java.util.Date;
@@ -59,18 +63,28 @@
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 
+import static com.aws.greengrass.componentmanager.KernelConfigResolver.CONFIGURATION_CONFIG_KEY;
 import static com.aws.greengrass.deployment.converter.DeploymentDocumentConverter.LOCAL_DEPLOYMENT_GROUP_NAME;
 import static com.aws.greengrass.integrationtests.logmanager.util.LogFileHelper.DEFAULT_FILE_SIZE;
 import static com.aws.greengrass.integrationtests.logmanager.util.LogFileHelper.DEFAULT_LOG_LINE_IN_FILE;
 import static com.aws.greengrass.integrationtests.logmanager.util.LogFileHelper.createFileAndWriteData;
 import static com.aws.greengrass.integrationtests.logmanager.util.LogFileHelper.createTempFileAndWriteData;
 import static com.aws.greengrass.logging.impl.config.LogConfig.newLogConfigFromRootConfig;
 import static com.aws.greengrass.logmanager.CloudWatchAttemptLogsProcessor.DEFAULT_LOG_STREAM_NAME;
+import static com.aws.greengrass.logmanager.LogManagerService.COMPONENT_LOGS_CONFIG_MAP_TOPIC_NAME;
 import static com.aws.greengrass.logmanager.LogManagerService.DEFAULT_FILE_REGEX;
+import static com.aws.greengrass.logmanager.LogManagerService.LOGS_UPLOADER_CONFIGURATION_TOPIC;
+import static com.aws.greengrass.logmanager.LogManagerService.MIN_LOG_LEVEL_CONFIG_TOPIC_NAME;
+import static com.aws.greengrass.logmanager.LogManagerService.PERSISTED_COMPONENT_CURRENT_PROCESSING_FILE_INFORMATION_V2;
 import static com.aws.greengrass.testcommons.testutilities.ExceptionLogProtector.ignoreExceptionOfType;
 import static com.aws.greengrass.testcommons.testutilities.ExceptionLogProtector.ignoreExceptionWithMessage;
+import static com.github.grantwest.eventually.EventuallyLambdaMatcher.eventuallyEval;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.atLeastOnce;
@@ -138,7 +152,8 @@ void setupKernel(Path storeDirectory, String configFileName) throws InterruptedE
                 logManagerService = (LogManagerService) service;
             }
         });
-        deviceConfiguration = new DeviceConfiguration(kernel, "ThingName", "xxxxxx-ats.iot.us-east-1.amazonaws.com", "xxxxxx.credentials.iot.us-east-1.amazonaws.com", "privKeyFilePath",
+        deviceConfiguration = new DeviceConfiguration(kernel, "ThingName", "xxxxxx-ats.iot.us-east-1.amazonaws.com",
+                "xxxxxx.credentials.iot.us-east-1.amazonaws.com", "privKeyFilePath",
                 "certFilePath", "caFilePath", "us-east-1", "roleAliasName");
 
         kernel.getContext().put(DeviceConfiguration.class, deviceConfiguration);
@@ -337,7 +352,167 @@ void GIVEN_log_manager_in_errored_state_WHEN_restarted_THEN_logs_upload_is_reatt
         assertEquals(1, logFileGroup.getLogFiles().size());
     }
 
-    private Runnable subscribeToActiveFileProcessed(LogManagerService service, int waitTime) throws InterruptedException {
+    @Test
+    @Tag("processingFilesInformation")
+    void GIVEN_filesDeletedAfterUpload_THEN_deletedFilesRemovedFromCache() throws Exception {
+        // Given
+
+        when(cloudWatchLogsClient.putLogEvents(any(PutLogEventsRequest.class)))
+                .thenReturn(PutLogEventsResponse.builder().nextSequenceToken("nextToken").build());
+
+        int numberOfFiles = 100;
+        tempDirectoryPath = Files.createTempDirectory(tempRootDir, "IntegrationTestsTemporaryLogFiles");
+        for (int i = 0; i < numberOfFiles; i++) {
+            createTempFileAndWriteData(tempDirectoryPath, "integTestRandomLogFiles.log_", "");
+        }
+
+        // When
+
+        String componentName = "UserComponentA";
+        // This configuration deletes files after upload
+        setupKernel(tempDirectoryPath, "smallPeriodicIntervalUserComponentConfig.yaml");
+
+        Runnable waitForActiveFileToBeProcessed = subscribeToActiveFileProcessed(logManagerService, 30);
+        waitForActiveFileToBeProcessed.run();
+        verify(cloudWatchLogsClient, atLeastOnce()).putLogEvents(captor.capture());
+
+        // Then
+
+        ProcessingFiles processingFiles = logManagerService.processingFilesInformation.get(componentName);
+        assertNotNull(processingFiles);
+        assertEquals(1, processingFiles.size()); // Active file not deleted
+
+        // Check runtime config gets cleared once the files have deleted
+        Topics componentTopics =
+                logManagerService.getRuntimeConfig()
+                        .lookupTopics(PERSISTED_COMPONENT_CURRENT_PROCESSING_FILE_INFORMATION_V2,
+                                componentName);
+        assertEquals(1, componentTopics.size());
+    }
+
+    @Test
+    @Tag("processingFilesInformation")
+    void GIVEN_filesNOTDeletedAfterUpload_THEN_filesGetCached() throws Exception {
+        // Given
+
+        when(cloudWatchLogsClient.putLogEvents(any(PutLogEventsRequest.class)))
+                .thenReturn(PutLogEventsResponse.builder().nextSequenceToken("nextToken").build());
+
+        int numberOfFiles = 100;
+        tempDirectoryPath = Files.createTempDirectory(tempRootDir, "IntegrationTestsTemporaryLogFiles");
+        for (int i = 0; i < numberOfFiles; i++) {
+            createTempFileAndWriteData(tempDirectoryPath, "integTestRandomLogFiles.log_", "");
+        }
+
+        // When
+        String componentName = "UserComponentA";
+        setupKernel(tempDirectoryPath, "doNotDeleteFilesAfterUpload.yaml");
+
+        Runnable waitForActiveFileToBeProcessed = subscribeToActiveFileProcessed(logManagerService, 30);
+        waitForActiveFileToBeProcessed.run();
+        verify(cloudWatchLogsClient, atLeastOnce()).putLogEvents(captor.capture());
+
+        // Then
+
+        // Note we shouldn't be accessing methods like this. Refactor this tests later
+        ProcessingFiles processingFiles = logManagerService.processingFilesInformation.get(componentName);
+        assertNotNull(processingFiles);
+        assertEquals(numberOfFiles, processingFiles.size());
+
+        // Check runtime config gets cleared once the files have deleted
+        Topics componentTopics =
+                logManagerService.getRuntimeConfig()
+                        .lookupTopics(PERSISTED_COMPONENT_CURRENT_PROCESSING_FILE_INFORMATION_V2,
+                                componentName);
+        assertEquals(numberOfFiles, componentTopics.size());
+    }
+
+    @Test
+    @Tag("processingFilesInformation")
+    void GIVEN_filesNOTDeletedAfterUpload_WHEN_removingComponentConfigurationNob_THEN_filesRemovedFromCache() throws
+            Exception {
+        // Given
+        when(cloudWatchLogsClient.putLogEvents(any(PutLogEventsRequest.class)))
+                .thenReturn(PutLogEventsResponse.builder().nextSequenceToken("nextToken").build());
+
+        int numberOfFiles = 100;
+        tempDirectoryPath = Files.createTempDirectory(tempRootDir, "IntegrationTestsTemporaryLogFiles");
+        for (int i = 0; i < numberOfFiles; i++) {
+            createTempFileAndWriteData(tempDirectoryPath, "integTestRandomLogFiles.log_", "");
+        }
+
+        // When
+
+        String componentName = "UserComponentA";
+        setupKernel(tempDirectoryPath, "doNotDeleteFilesAfterUpload.yaml");
+
+        Runnable waitForActiveFileToBeProcessed = subscribeToActiveFileProcessed(logManagerService, 30);
+        waitForActiveFileToBeProcessed.run();
+        verify(cloudWatchLogsClient, atLeastOnce()).putLogEvents(captor.capture());
+
+        // Component configuration is removed
+
+        logManagerService.getConfig().lookupTopics(CONFIGURATION_CONFIG_KEY, LOGS_UPLOADER_CONFIGURATION_TOPIC,
+                COMPONENT_LOGS_CONFIG_MAP_TOPIC_NAME, componentName).remove();
+
+        assertThat(() -> logManagerService.getComponentLogConfigurations().get(componentName) == null,
+                eventuallyEval(equalTo(true), Duration.ofSeconds(30)));
+
+        // Then
+
+        ProcessingFiles processingFiles = logManagerService.processingFilesInformation.get(componentName);
+        assertNull(processingFiles);
+
+        // Check runtime config gets cleared once the files have deleted
+        Topics componentTopics =
+                logManagerService.getRuntimeConfig()
+                        .lookupTopics(PERSISTED_COMPONENT_CURRENT_PROCESSING_FILE_INFORMATION_V2,
+                                componentName);
+        assertEquals(0, componentTopics.size());
+    }
+
+    @Test
+    @Tag("processingFilesInformation")
+    void GIVEN_processingFileCached_WHEN_ConfigurationChanges_THEN_existingCachedFileInformationLastAccessedIsNotChanged() throws
+            Exception {
+        // Given
+        when(cloudWatchLogsClient.putLogEvents(any(PutLogEventsRequest.class)))
+                .thenReturn(PutLogEventsResponse.builder().nextSequenceToken("nextToken").build());
+
+        int numberOfFiles = 10;
+        tempDirectoryPath = Files.createTempDirectory(tempRootDir, "IntegrationTestsTemporaryLogFiles");
+        for (int i = 0; i < numberOfFiles; i++) {
+            createTempFileAndWriteData(tempDirectoryPath, "integTestRandomLogFiles.log_", "");
+        }
+
+        // When
+
+        String componentName = "UserComponentA";
+        setupKernel(tempDirectoryPath, "doNotDeleteFilesAfterUpload.yaml");
+
+        Runnable waitForActiveFileToBeProcessed = subscribeToActiveFileProcessed(logManagerService, 30);
+        waitForActiveFileToBeProcessed.run();
+        verify(cloudWatchLogsClient, atLeastOnce()).putLogEvents(captor.capture());
+
+        ProcessingFiles processingFilesBefore = logManagerService.processingFilesInformation.get(componentName);
+        Map<String, Object> beforeConfigurationUpdate = processingFilesBefore.toMap();
+
+        logManagerService.getConfig().lookup(CONFIGURATION_CONFIG_KEY, LOGS_UPLOADER_CONFIGURATION_TOPIC,
+                COMPONENT_LOGS_CONFIG_MAP_TOPIC_NAME, componentName,  MIN_LOG_LEVEL_CONFIG_TOPIC_NAME)
+                .withValue("WARN");
+
+        assertThat(()-> logManagerService.getComponentLogConfigurations().get(componentName).getMinimumLogLevel(),
+                eventuallyEval(is(Level.WARN), Duration.ofSeconds(30)));
+
+        // Then
+
+        ProcessingFiles processingFilesAfter = logManagerService.processingFilesInformation.get(componentName);
+        Map<String, Object> afterConfigurationUpdate = processingFilesAfter.toMap();
+        assertEquals(beforeConfigurationUpdate, afterConfigurationUpdate);
+    }
+
+    private Runnable subscribeToActiveFileProcessed(LogManagerService service, int waitTime) throws
+            InterruptedException {
         CountDownLatch latch = new CountDownLatch(1);
         service.registerEventStatusListener((EventType event) -> {
             if (event == EventType.ALL_COMPONENTS_PROCESSED) {

src/integrationtests/java/com/aws/greengrass/integrationtests/logmanager/SpaceManagementTest.java

@@ -180,6 +180,9 @@ void GIVEN_user_component_config_with_space_management_WHEN_space_exceeds_THEN_e
             throws Exception {
         // Given
 
+        lenient().when(cloudWatchLogsClient.putLogEvents(any(PutLogEventsRequest.class)))
+                .thenReturn(PutLogEventsResponse.builder().nextSequenceToken("nextToken").build());
+
         tempDirectoryPath = Files.createDirectory(tempRootDir.resolve("IntegrationTestsTemporaryLogFiles"));
         // This method configures the LogManager to get logs with the pattern ^integTestRandomLogFiles.log\w* inside
         // then tempDirectoryPath with a diskSpaceLimit of 105kb

src/integrationtests/resources/com/aws/greengrass/integrationtests/logmanager/doNotDeleteFilesAfterUpload.yaml

@@ -0,0 +1,18 @@
+---
+services:
+  aws.greengrass.LogManager:
+    configuration:
+      periodicUploadIntervalSec: 10
+      logsUploaderConfiguration:
+        componentLogsConfigurationMap:
+          UserComponentA:
+            logFileRegex: '^integTestRandomLogFiles.log\w*'
+            logFileDirectoryPath: '{{logFileDirectoryPath}}'
+            deleteLogFileAfterCloudUpload: 'false'
+
+  main:
+    lifecycle:
+      install:
+        all: echo All installed
+    dependencies:
+      - aws.greengrass.LogManager