Merge branch 'main' into 2024-12-06_main_add-time-sync

Author: junfuchen99

pom.xml

@@ -238,7 +238,7 @@
             When updating the version here, ensure you match the correct aws-crt version below.
             Get the correct version from: https://github.com/aws/aws-iot-device-sdk-java-v2/blob/main/sdk/pom.xml#L45
             !-->
-            <version>1.21.0</version>
+            <version>1.23.0</version>
             <exclusions>
                 <exclusion>
                     <groupId>software.amazon.awssdk.crt</groupId>
@@ -249,7 +249,7 @@
         <dependency>
             <groupId>software.amazon.awssdk.crt</groupId>
             <artifactId>aws-crt</artifactId>
-            <version>0.30.8</version>
+            <version>0.33.5</version>
             <classifier>fips-where-available</classifier>
         </dependency>
         <dependency>

src/main/java/com/aws/greengrass/builtin/services/pubsub/SubscriptionTrie.java

@@ -5,11 +5,14 @@
 
 package com.aws.greengrass.builtin.services.pubsub;
 
+import com.aws.greengrass.util.Utils;
+
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Trie to manage subscriptions.
@@ -69,11 +72,43 @@ public boolean remove(String topic, K cb) {
      * @return if changed after removal
      */
     public boolean remove(String topic, Set<K> cbs) {
-        SubscriptionTrie<K> sub = lookup(topic);
-        if (sub == null) {
+        if (Utils.isEmpty(topic) || Utils.isEmpty(cbs)) {
             return false;
         }
-        return sub.subscriptionCallbacks.removeAll(cbs);
+
+        AtomicBoolean isSubscriptionRemoved = new AtomicBoolean(false);
+        removeRecursively(topic.split(TOPIC_LEVEL_SEPARATOR), this, cbs, 0, isSubscriptionRemoved);
+        return isSubscriptionRemoved.get();
+    }
+
+    private boolean canRemove(SubscriptionTrie<K> node) {
+        // returns false if the topic level is prefix of another topic or has callbacks registered
+        return node.children.isEmpty() && node.subscriptionCallbacks.isEmpty();
+    }
+
+    /*
+    This method removes the requested callback from a topic and prunes the subscription trie recursively by
+    1. navigating to the last node of the requested topic
+    1.a. at this node, the method persists the result of the requested callback removal for the penultimate result
+    1.b. returns true if requested callback was removed and if current topic node can be pruned
+
+    2. the previous topic node receives the result from 1.b.
+    2.a. if true, remove child topic node and return if current node can be pruned
+    2.b. if false, simply do nothing and return false implying current node cannot be pruned
+     */
+    private boolean removeRecursively(String[] topicNodes, SubscriptionTrie<K> topicNode, Set<K> cbs, int index,
+                                      AtomicBoolean subscriptionRemoved) {
+        if (index == topicNodes.length) {
+            subscriptionRemoved.set(topicNode.subscriptionCallbacks.removeAll(cbs));
+            return subscriptionRemoved.get() && canRemove(topicNode);
+        }
+
+        if (removeRecursively(topicNodes, topicNode.children.get(topicNodes[index]), cbs, index + 1,
+                subscriptionRemoved)) {
+            topicNode.children.remove(topicNodes[index]);
+            return canRemove(topicNode);
+        }
+        return false;
     }
 
     /**
@@ -183,3 +218,4 @@ public static boolean isWildcard(String topic) {
     }
 
 }
+

src/main/java/com/aws/greengrass/componentmanager/ComponentManager.java

@@ -61,6 +61,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -71,6 +72,7 @@
 
 import static com.aws.greengrass.componentmanager.KernelConfigResolver.PREV_VERSION_CONFIG_KEY;
 import static com.aws.greengrass.componentmanager.KernelConfigResolver.VERSION_CONFIG_KEY;
+import static com.aws.greengrass.deployment.DeviceConfiguration.DEFAULT_NUCLEUS_COMPONENT_NAME;
 import static com.aws.greengrass.deployment.converter.DeploymentDocumentConverter.ANY_VERSION;
 import static org.apache.commons.io.FileUtils.ONE_MB;
 
@@ -296,6 +298,37 @@ private void storeRecipeDigestInConfigStoreForPlugin(
         }
     }
 
+    /**
+     * Un-archives the artifacts for the current Nucleus version package.
+     *
+     * @return list of un-archived paths
+     * @throws PackageLoadingException when unable to load current Nucleus
+     */
+    public List<Path> unArchiveCurrentNucleusVersionArtifacts() throws PackageLoadingException {
+        String currentNucleusVersion = deviceConfiguration.getNucleusVersion();
+        ComponentIdentifier nucleusComponentIdentifier =
+                new ComponentIdentifier(DEFAULT_NUCLEUS_COMPONENT_NAME, new Semver(currentNucleusVersion));
+        List<File> nucleusArtifactFileNames =
+                componentStore.getArtifactFiles(nucleusComponentIdentifier, artifactDownloaderFactory);
+        return nucleusArtifactFileNames.stream()
+                .map(file -> {
+                    try {
+                        Path unarchivePath =
+                                nucleusPaths.unarchiveArtifactPath(nucleusComponentIdentifier, getFileName(file));
+                        /*
+                        Using a hard-coded ZIP un-archiver as today this code path is only used to un-archive a Nucleus
+                        .zip artifact.
+                         */
+                        unarchiver.unarchive(Unarchive.ZIP, file, unarchivePath);
+                        return unarchivePath;
+                    } catch (IOException e) {
+                        logger.atDebug().setCause(e).kv("comp-id", nucleusComponentIdentifier)
+                                .log("Could not un-archive Nucleus artifact");
+                        return null;
+                    }
+                }).filter(Objects::nonNull).collect(Collectors.toList());
+    }
+
     private Optional<ComponentIdentifier> findBestCandidateLocally(String componentName,
                                                                    Map<String, Requirement> versionRequirements)
             throws PackagingException {

src/main/java/com/aws/greengrass/componentmanager/ComponentStore.java

@@ -47,8 +47,10 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.stream.Collectors;
 import java.util.stream.LongStream;
 import javax.inject.Inject;
 
@@ -412,6 +414,37 @@ public Path resolveArtifactDirectoryPath(@NonNull ComponentIdentifier componentI
         }
     }
 
+    /**
+     * Returns the artifact file name.
+     *
+     * @param componentIdentifier packageIdentifier
+     * @param artifactDownloaderFactory artifact downloader factory
+     * @return the unarchive artifact directory path for target package.
+     * @throws PackageLoadingException if creating the directory fails
+     */
+    public List<File> getArtifactFiles(@NonNull ComponentIdentifier componentIdentifier,
+                                       @NonNull ArtifactDownloaderFactory artifactDownloaderFactory)
+            throws PackageLoadingException {
+        Optional<String> componentRecipeContent = findComponentRecipeContent(componentIdentifier);
+        if (!componentRecipeContent.isPresent()) {
+            return Collections.emptyList();
+        }
+
+        ComponentRecipe recipe = getPackageRecipe(componentIdentifier);
+        Path packageArtifactDirectory = resolveArtifactDirectoryPath(componentIdentifier);
+        return recipe.getArtifacts().stream().map(artifact -> {
+            try {
+                return artifactDownloaderFactory
+                        .getArtifactDownloader(componentIdentifier, artifact, packageArtifactDirectory)
+                        .getArtifactFile();
+            } catch (PackageLoadingException | InvalidArtifactUriException e) {
+                logger.atDebug().setCause(e).kv("comp-id", componentRecipeContent)
+                        .log("Could not get artifact file");
+                return null;
+            }
+        }).filter(Objects::nonNull).collect(Collectors.toList());
+    }
+
     /**
      * Resolve the recipe file path for a target package id.
      *

src/main/java/com/aws/greengrass/deployment/errorcode/DeploymentErrorCode.java

@@ -64,7 +64,7 @@ public enum DeploymentErrorCode {
     // JVM hashing issue
     HASHING_ALGORITHM_UNAVAILABLE(DeploymentErrorType.DEVICE_ERROR),
     // Could be a local file issue or a Nucleus issue; we will categorize as the latter for visibility
-    LAUNCH_DIRECTORY_CORRUPTED(DeploymentErrorType.NUCLEUS_ERROR),
+    LAUNCH_DIRECTORY_CORRUPTED(DeploymentErrorType.DEVICE_ERROR),
 
     /* Component recipe errors */
     RECIPE_PARSE_ERROR(DeploymentErrorType.COMPONENT_RECIPE_ERROR),

src/main/java/com/aws/greengrass/lifecyclemanager/KernelAlternatives.java

@@ -5,6 +5,8 @@
 
 package com.aws.greengrass.lifecyclemanager;
 
+import com.aws.greengrass.componentmanager.ComponentManager;
+import com.aws.greengrass.componentmanager.exceptions.PackageLoadingException;
 import com.aws.greengrass.config.Configuration;
 import com.aws.greengrass.config.Topics;
 import com.aws.greengrass.dependency.Context;
@@ -35,6 +37,8 @@
 import java.nio.file.Path;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import javax.inject.Inject;
 
@@ -70,15 +74,18 @@ public class KernelAlternatives {
     private static final String BOOTSTRAP_ON_ROLLBACK_CONFIG_KEY = "bootstrapOnRollback";
 
     private final NucleusPaths nucleusPaths;
+    private final ComponentManager componentManager;
 
     /**
      * Constructor for KernelAlternatives, which manages the alternative launch directory of Kernel.
      *
      * @param nucleusPaths nucleus paths
+     * @param componentManager component manager
      */
     @Inject
-    public KernelAlternatives(NucleusPaths nucleusPaths) {
+    public KernelAlternatives(NucleusPaths nucleusPaths, ComponentManager componentManager) {
         this.nucleusPaths = nucleusPaths;
+        this.componentManager = componentManager;
         try {
             setupInitLaunchDirIfAbsent();
         } catch (IOException e) {
@@ -162,31 +169,69 @@ public boolean isLaunchDirSetup() {
         return Files.isSymbolicLink(getCurrentDir()) && validateLaunchDirSetup(getCurrentDir());
     }
 
+    protected boolean canRecoverMissingLaunchDirSetup()
+            throws IOException, URISyntaxException, PackageLoadingException {
+        /*
+        Try and relink launch dir with the following replacement criteria
+        1. check if current Nucleus execution package is valid
+        2. un-archive current Nucleus version from component store
+        3. fail with DirectoryValidationException if above steps do not satisfy
+         */
+        Path currentNucleusExecutablePath = locateCurrentKernelUnpackDir();
+        if (Files.exists(currentNucleusExecutablePath.resolve(KERNEL_BIN_DIR)
+                .resolve(Platform.getInstance().loaderFilename()))) {
+            logger.atDebug().kv("path", currentNucleusExecutablePath)
+                    .log("Current Nucleus executable is valid, setting up launch dir");
+            relinkInitLaunchDir(currentNucleusExecutablePath, true);
+            return true;
+        }
+
+        List<Path> localNucleusExecutablePaths = componentManager.unArchiveCurrentNucleusVersionArtifacts();
+        if (!localNucleusExecutablePaths.isEmpty()) {
+            Optional<Path> validNucleusExecutablePath = localNucleusExecutablePaths.stream()
+                    .filter(path -> Files.exists(path.resolve(KERNEL_BIN_DIR)
+                            .resolve(Platform.getInstance().loaderFilename())))
+                    .findFirst();
+            if (validNucleusExecutablePath.isPresent()) {
+                logger.atDebug().kv("path", validNucleusExecutablePath.get())
+                        .log("Un-archived current Nucleus artifact");
+                relinkInitLaunchDir(validNucleusExecutablePath.get(), true);
+                return true;
+            }
+        }
+        throw new PackageLoadingException("Could not find a valid Nucleus package to recover launch dir setup");
+    }
+
     /**
      * Validate that launch directory is set up.
      *
      * @throws DirectoryValidationException when a file is missing
      * @throws DeploymentException when user is not allowed to change file permission
      */
     public void validateLaunchDirSetupVerbose() throws DirectoryValidationException, DeploymentException {
-        Path currentDir = getCurrentDir();
-        if (!Files.isSymbolicLink(currentDir)) {
-            throw new DirectoryValidationException("Missing symlink to current nucleus launch directory");
+        try {
+            if (!Files.isSymbolicLink(getCurrentDir()) || !Files.exists(getLoaderPathFromLaunchDir(getCurrentDir()))) {
+                logger.atInfo().log("Current launch dir setup is missing, attempting to recover");
+                canRecoverMissingLaunchDirSetup();
+            }
+        } catch (PackageLoadingException | IOException ex) {
+            throw new DirectoryValidationException("Unable to relink init launch directory", ex);
+        } catch (URISyntaxException ex) {
+            // TODO: Fix usage of root path with spaces on linux
+            throw new DeploymentException("Could not parse init launch directory path", ex);
         }
+
+        Path currentDir = getCurrentDir();
         Path loaderPath = getLoaderPathFromLaunchDir(currentDir);
-        if (Files.exists(loaderPath)) {
-            if (!loaderPath.toFile().canExecute()) {
-                // Ensure that the loader is executable so that we can exec it when restarting Nucleus
-                try {
-                    Platform.getInstance().setPermissions(OWNER_RWX_EVERYONE_RX, loaderPath);
-                } catch (IOException e) {
-                    throw new DeploymentException(
-                            String.format("Unable to set loader script at %s as executable", loaderPath), e)
-                            .withErrorContext(e, DeploymentErrorCode.SET_PERMISSION_ERROR);
-                }
+        if (!loaderPath.toFile().canExecute()) {
+            // Ensure that the loader is executable so that we can exec it when restarting Nucleus
+            try {
+                Platform.getInstance().setPermissions(OWNER_RWX_EVERYONE_RX, loaderPath);
+            } catch (IOException e) {
+                throw new DeploymentException(
+                        String.format("Unable to set loader script at %s as executable", loaderPath), e)
+                        .withErrorContext(e, DeploymentErrorCode.SET_PERMISSION_ERROR);
             }
-        } else {
-            throw new DirectoryValidationException("Missing loader file at " + currentDir.toAbsolutePath());
         }
     }
 

src/main/java/com/aws/greengrass/lifecyclemanager/exceptions/DirectoryValidationException.java

@@ -15,4 +15,9 @@ public DirectoryValidationException(String message) {
         super(message);
         super.addErrorCode(DeploymentErrorCode.LAUNCH_DIRECTORY_CORRUPTED);
     }
+
+    public DirectoryValidationException(String message, Throwable throwable) {
+        super(message, throwable);
+        super.addErrorCode(DeploymentErrorCode.LAUNCH_DIRECTORY_CORRUPTED);
+    }
 }

src/test/java/com/aws/greengrass/builtin/services/pubsub/SubscriptionTrieTest.java

@@ -137,6 +137,51 @@ public void GIVEN_subscription_wildcard_WHEN_remove_topic_THEN_no_matches() {
         assertEquals(0, trie.size());
     }
 
+    @Test
+    public void GIVEN_subscriptions_with_wildcards_WHEN_remove_topics_THEN_clean_up_trie() {
+        assertEquals(0, trie.size());
+        SubscriptionCallback cb1 = generateSubscriptionCallback();
+        SubscriptionCallback cb2 = generateSubscriptionCallback();
+        SubscriptionCallback cb3 = generateSubscriptionCallback();
+        String topic = "foo/+/bar/#";
+        trie.add("bar", cb1);
+        trie.add(topic, cb1);
+        trie.add(topic, cb2);
+        // Topic is not registered with the callback, mark it as removed when requested to remove
+        assertThat("remove topic", trie.remove(topic, cb3), is(false));
+        assertThat(trie.get(topic), containsInAnyOrder(cb1, cb2));
+
+        trie.add("foo/#", cb3);
+        trie.add("foo/+", cb2);
+
+        assertThat(trie.get(topic), containsInAnyOrder(cb1, cb2, cb3));
+        assertEquals(5, trie.size());
+
+        assertThat("remove topic", trie.remove("foo/+", cb2), is(true));
+        assertEquals(4, trie.size());
+        assertThat(trie.get("foo/+"), containsInAnyOrder(cb3)); // foo/+ still matches with foo/#
+        assertThat(trie.get(topic), containsInAnyOrder(cb1, cb2, cb3)); // foo/+/bar/# still exists
+
+        assertThat("remove topic", trie.remove("foo/#", cb3), is(true));
+        assertFalse(trie.containsKey("foo/#"));
+        assertThat(trie.get("foo/#"), is(empty()));
+        assertThat(trie.get("foo/+"), is(empty())); // foo/+ doesn't match with any existing topic
+        assertThat(trie.get(topic), containsInAnyOrder(cb1, cb2)); // foo/+/bar/# still exists
+        assertEquals(3, trie.size());
+
+        assertThat("remove topic", trie.remove(topic, cb1), is(true));
+        assertThat(trie.get(topic), contains(cb2));
+        assertEquals(2, trie.size());
+        assertTrue(trie.containsKey("foo/+"));
+        assertTrue(trie.containsKey("foo/+/bar/#"));
+
+        assertThat("remove topic", trie.remove(topic, cb2), is(true));
+        assertThat(trie.get(topic), is(empty()));
+        assertEquals(1, trie.size());
+        assertFalse(trie.containsKey("foo/+"));
+        assertFalse(trie.containsKey("foo/+/bar/#"));
+    }
+
     @Test
     void GIVEN_topics_WHEN_isWildcard_THEN_returns_whether_it_uses_wildcard() {
         assertTrue(SubscriptionTrie.isWildcard("+"));

src/test/java/com/aws/greengrass/deployment/activator/KernelUpdateActivatorTest.java

@@ -227,7 +227,7 @@ void GIVEN_launch_dir_corrupted_WHEN_deployment_activate_THEN_deployment_fail(Ex
         assertEquals(mockException, result.getFailureCause().getCause());
 
         List<String> expectedStack = Arrays.asList("DEPLOYMENT_FAILURE", "LAUNCH_DIRECTORY_CORRUPTED");
-        List<String> expectedTypes = Collections.singletonList("NUCLEUS_ERROR");
+        List<String> expectedTypes = Collections.singletonList("DEVICE_ERROR");
         TestUtils.validateGenerateErrorReport(result.getFailureCause(), expectedStack, expectedTypes);
     }