Added support for zipping fragment content in k8s (#1047)

Author: codepitbull

hivemq-edge/src/main/java/com/hivemq/api/model/adapters/AdapterStatusModelConversionUtils.java

@@ -47,29 +47,20 @@ public class AdapterStatusModelConversionUtils {
 
     public static @NotNull Status.ConnectionEnum convertConnectionStatus(final @NotNull ProtocolAdapterState.ConnectionStatus connectionStatus) {
         Preconditions.checkNotNull(connectionStatus);
-        switch (connectionStatus) {
-            case DISCONNECTED:
-                return Status.ConnectionEnum.DISCONNECTED;
-            case CONNECTED:
-                return Status.ConnectionEnum.CONNECTED;
-            case ERROR:
-                return Status.ConnectionEnum.ERROR;
-            case STATELESS:
-                return Status.ConnectionEnum.STATELESS;
-            default:
-            case UNKNOWN:
-                return Status.ConnectionEnum.UNKNOWN;
-        }
+        return switch (connectionStatus) {
+            case DISCONNECTED -> Status.ConnectionEnum.DISCONNECTED;
+            case CONNECTED -> Status.ConnectionEnum.CONNECTED;
+            case ERROR -> Status.ConnectionEnum.ERROR;
+            case STATELESS -> Status.ConnectionEnum.STATELESS;
+            default -> Status.ConnectionEnum.UNKNOWN;
+        };
     }
 
     public static @NotNull Status.RuntimeEnum convertRuntimeStatus(final @NotNull ProtocolAdapterState.RuntimeStatus runtimeStatus) {
         Preconditions.checkNotNull(runtimeStatus);
-        switch (runtimeStatus) {
-            case STARTED:
-                return Status.RuntimeEnum.STARTED;
-            default:
-            case STOPPED:
-                return Status.RuntimeEnum.STOPPED;
+        if(ProtocolAdapterState.RuntimeStatus.STARTED.equals(runtimeStatus)) {
+            return Status.RuntimeEnum.STARTED;
         }
+        return Status.RuntimeEnum.STOPPED;
     }
 }

hivemq-edge/src/main/java/com/hivemq/configuration/ConfigurationBootstrap.java

@@ -70,7 +70,9 @@ public class ConfigurationBootstrap {
 
         final ConfigurationFile configurationFile = ConfigurationFileProvider.get(systemInformation);
 
-        final ConfigFileReaderWriter configFileReader = new ConfigFileReaderWriter(configurationFile,
+        final ConfigFileReaderWriter configFileReader = new ConfigFileReaderWriter(
+                systemInformation,
+                configurationFile,
                 List.of(
                         new RestrictionConfigurator(configurationService.restrictionsConfiguration()),
                         new SecurityConfigurator(configurationService.securityConfiguration()),

hivemq-edge/src/main/java/com/hivemq/configuration/EnvironmentVariables.java

@@ -58,6 +58,11 @@ public class EnvironmentVariables {
      */
     public static final String CONFIG_WRITEABLE = "HIVEMQ_CONFIG_WRITEABLE";
 
+    /**
+     * Name of the environment variable for indicating that the config fragment will be zipped and base64 encoded
+     */
+    public static final String CONFIG_FRAGMENT_BASE64ZIP = "HIVEMQ_CONFIG_FRAGMENT_BASE64ZIP";
+
     /**
      * Name of the environment variable for configuring the data folder.
      */

hivemq-edge/src/main/java/com/hivemq/configuration/SystemProperties.java

@@ -28,6 +28,7 @@ public class SystemProperties {
     public static final String CONFIG_FOLDER_SECONDARY = "hivemq.config.secondary";
     public static final String CONFIG_REFRESH_INTERVAL = "hivemq.config.refreshinterval";
     public static final String CONFIG_WRITEABLE = "hivemq.config.writeable";
+    public static final String CONFIG_FRAGMENT_BASE64ZIP = "hivemq.config.fragment.base64zip";
     public static final String LICENSE_FOLDER = "hivemq.license.folder";
 
     public static final String DATA_FOLDER = "hivemq.data.folder";

hivemq-edge/src/main/java/com/hivemq/configuration/info/SystemInformation.java

@@ -104,6 +104,11 @@ public interface SystemInformation {
      */
     boolean isEmbedded();
 
+    /**
+     * @return should the fragment config be treated as bing zipped and base64 encoded
+     */
+    boolean isConfigFragmentBase64Zip();
+
     /**
      * @return the interval between refreshing config files, 0 means no refreshing
      */

hivemq-edge/src/main/java/com/hivemq/configuration/info/SystemInformationImpl.java

@@ -59,6 +59,8 @@ public class SystemInformationImpl implements SystemInformation {
 
     private final boolean configWriteable;
 
+    private final boolean configFragmentBase64Zip;
+
     public SystemInformationImpl() {
         this(false);
     }
@@ -78,6 +80,7 @@ public SystemInformationImpl(
             final @Nullable File licenseFolder) {
         final String refreshInterval = getSystemPropertyOrEnvironmentVariable(SystemProperties.CONFIG_REFRESH_INTERVAL, EnvironmentVariables.CONFIG_REFRESH_INTERVAL);
         final String configWriteable = getSystemPropertyOrEnvironmentVariable(SystemProperties.CONFIG_WRITEABLE, EnvironmentVariables.CONFIG_WRITEABLE);
+        final String configFragmentBase64ZipString = getSystemPropertyOrEnvironmentVariable(SystemProperties.CONFIG_FRAGMENT_BASE64ZIP, EnvironmentVariables.CONFIG_FRAGMENT_BASE64ZIP);
         this.usePathOfRunningJar = usePathOfRunningJar;
         this.embedded = embedded;
         this.configFolder = configFolder;
@@ -89,6 +92,7 @@ public SystemInformationImpl(
         this.runningSince = System.currentTimeMillis();
         this.configRefreshIntervalInMs = Long.parseLong(Objects.requireNonNullElse(refreshInterval, "-1"));
         this.configWriteable = Boolean.parseBoolean((configWriteable == null || configWriteable.isEmpty()) ? "true" : configWriteable );
+        this.configFragmentBase64Zip = Boolean.parseBoolean((configFragmentBase64ZipString == null || configFragmentBase64ZipString.isEmpty()) ? "false" : configFragmentBase64ZipString );
         processorCount = getPhysicalProcessorCount();
     }
 
@@ -366,4 +370,9 @@ public boolean isEmbedded() {
     public boolean isConfigWriteable() {
         return configWriteable;
     }
+
+    @Override
+    public boolean isConfigFragmentBase64Zip() {
+        return configFragmentBase64Zip;
+    }
 }

hivemq-edge/src/main/java/com/hivemq/configuration/reader/ConfigFileReaderWriter.java

@@ -29,12 +29,20 @@
 import com.hivemq.configuration.entity.listener.WebsocketListenerEntity;
 import com.hivemq.configuration.entity.listener.tls.KeystoreEntity;
 import com.hivemq.configuration.entity.listener.tls.TruststoreEntity;
+import com.hivemq.configuration.info.SystemInformation;
 import com.hivemq.edge.HiveMQEdgeConstants;
 import com.hivemq.exceptions.UnrecoverableException;
 import com.hivemq.util.ThreadFactoryUtil;
 import com.hivemq.util.render.EnvVarUtil;
 import com.hivemq.util.render.FileFragmentUtil;
 import com.hivemq.util.render.IfUtil;
+import jakarta.xml.bind.JAXBContext;
+import jakarta.xml.bind.JAXBElement;
+import jakarta.xml.bind.JAXBException;
+import jakarta.xml.bind.Marshaller;
+import jakarta.xml.bind.Unmarshaller;
+import jakarta.xml.bind.ValidationEvent;
+import jakarta.xml.bind.ValidationEventLocator;
 import org.apache.commons.io.FileUtils;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -44,17 +52,11 @@
 import org.xml.sax.SAXException;
 
 import javax.xml.XMLConstants;
-import jakarta.xml.bind.JAXBContext;
-import jakarta.xml.bind.JAXBElement;
-import jakarta.xml.bind.JAXBException;
-import jakarta.xml.bind.Marshaller;
-import jakarta.xml.bind.Unmarshaller;
-import jakarta.xml.bind.ValidationEvent;
-import jakarta.xml.bind.ValidationEventLocator;
 import javax.xml.transform.stream.StreamSource;
 import javax.xml.validation.Schema;
 import javax.xml.validation.SchemaFactory;
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
@@ -74,13 +76,16 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
+import java.util.zip.DataFormatException;
+import java.util.zip.Inflater;
 
 import static java.util.Objects.requireNonNullElse;
 
@@ -91,7 +96,7 @@ public class ConfigFileReaderWriter {
     public static final String CONFIG_FRAGMENT_PATH = "/fragment/config";
 
     private final @NotNull ConfigurationFile configurationFile;
-    protected volatile @NotNull HiveMQConfigEntity configEntity;
+    protected volatile HiveMQConfigEntity configEntity;
     private final Object lock = new Object();
     private boolean defaultBackupConfig = true;
     private volatile @Nullable ScheduledExecutorService scheduledExecutorService = null;
@@ -103,10 +108,12 @@ public class ConfigFileReaderWriter {
     private final @NotNull DataCombiningExtractor dataCombiningExtractor;
     private final @NotNull UnsExtractor unsExtractor;
     private final @NotNull List<ReloadableExtractor> reloadableExtractors;
+    private final @NotNull SystemInformation systemInformation;
 
     private final @NotNull AtomicLong lastWrite = new AtomicLong(0L);
 
     public ConfigFileReaderWriter(
+            final @NotNull SystemInformation systemInformation,
             final @NotNull ConfigurationFile configurationFile,
             final @NotNull List<Configurator<?>> configurators) {
         this.configurationFile = configurationFile;
@@ -115,6 +122,7 @@ public ConfigFileReaderWriter(
         this.protocolAdapterExtractor = new ProtocolAdapterExtractor(this);
         this.dataCombiningExtractor = new DataCombiningExtractor(this);
         this.unsExtractor = new UnsExtractor(this);
+        this.systemInformation = systemInformation;
         reloadableExtractors = List.of(
                 bridgeExtractor,
                 protocolAdapterExtractor,
@@ -201,14 +209,17 @@ private void checkMonitoredFilesForChanges(
 
                 pathsToCheck.putAll(fileModificationTimestamps);
 
-                pathsToCheck.entrySet().forEach(pathToTs -> {
+                pathsToCheck.forEach((key, value) -> {
                     try {
-                        if (!pathToTs.getKey().toString().equals(CONFIG_FRAGMENT_PATH) && Files.getFileAttributeView(pathToTs.getKey().toRealPath(LinkOption.NOFOLLOW_LINKS), BasicFileAttributeView.class).readAttributes().lastModifiedTime().toMillis() > pathToTs.getValue()) {
-                            log.error("Restarting because a required file was updated: {}", pathToTs.getKey());
+                        if (!key.toString().equals(CONFIG_FRAGMENT_PATH) &&
+                                Files.getFileAttributeView(key.toRealPath(LinkOption.NOFOLLOW_LINKS),
+                                        BasicFileAttributeView.class).readAttributes().lastModifiedTime().toMillis() >
+                                        value) {
+                            log.error("Restarting because a required file was updated: {}", key);
                             System.exit(0);
                         }
                     } catch (final IOException e) {
-                        throw new RuntimeException("Unable to read last modified time for " + pathToTs.getKey(), e);
+                        throw new RuntimeException("Unable to read last modified time for " + key, e);
                     }
                 });
             }
@@ -339,8 +350,7 @@ protected JAXBContext createContext() throws JAXBException {
                 .build()
                 .toArray(new Class<?>[0]);
 
-        final JAXBContext context = JAXBContext.newInstance(classes);
-        return context;
+        return JAXBContext.newInstance(classes);
     }
 
     private void writeConfigToXML(final @NotNull ConfigurationFile outputFile, final boolean rollConfig) {
@@ -442,7 +452,10 @@ public void writeConfigToXML(final @NotNull Writer writer) {
 
                 //replace environment variable placeholders
                 String configFileContent = Files.readString(configFile.toPath());
-                final FileFragmentUtil.FragmentResult fragmentResult = FileFragmentUtil.replaceFragmentPlaceHolders(configFileContent);
+                final var fragmentResult = FileFragmentUtil
+                        .replaceFragmentPlaceHolders(
+                                configFileContent,
+                                systemInformation.isConfigFragmentBase64Zip());
 
                 fragmentToModificationTime.putAll(fragmentResult.getFragmentToModificationTime());
 
@@ -461,7 +474,6 @@ public void writeConfigToXML(final @NotNull Writer writer) {
                     return true;
 
                 });
-
                 final JAXBElement<? extends HiveMQConfigEntity> result =
                         unmarshaller.unmarshal(streamSource, getConfigEntityClass());
 
@@ -587,8 +599,7 @@ protected Schema loadSchema() throws IOException, SAXException {
         if (resource != null) {
             try (final InputStream is = uncachedStream(resource)) {
                 final SchemaFactory sf = SchemaFactory.newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI);
-                final Schema schema = sf.newSchema(new StreamSource(is));
-                return schema;
+                return sf.newSchema(new StreamSource(is));
             }
         }
         log.warn("No schema loaded for validation of config xml.");

hivemq-edge/src/main/java/com/hivemq/util/render/FileFragmentUtil.java

@@ -17,17 +17,25 @@
 
 import com.hivemq.exceptions.UnrecoverableException;
 import org.jetbrains.annotations.NotNull;
+import org.jose4j.base64url.Base64;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.zip.DataFormatException;
+import java.util.zip.Inflater;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
 
 public class FileFragmentUtil {
     private static final Logger log = LoggerFactory.getLogger(EnvVarUtil.class);
@@ -41,9 +49,9 @@ public class FileFragmentUtil {
      * @return replacement result containing the actual string and file and their modification time used in the replacement
      * @throws UnrecoverableException if a fragment used in a placeholder can't be loaded
      */
-    public static @NotNull FragmentResult replaceFragmentPlaceHolders(final @NotNull String text) {
+    public static @NotNull FragmentResult replaceFragmentPlaceHolders(final @NotNull String text, final boolean contentIsZippedAndBase64Encoded) {
 
-        final StringBuffer resultString = new StringBuffer();
+        final StringBuilder resultString = new StringBuilder();
         final Map<Path, Long> fragmentToModificationTime = new HashMap<>();
         final Matcher matcher = Pattern.compile(FRAGMENT_VAR_PATTERN)
                 .matcher(text);
@@ -60,8 +68,19 @@ public class FileFragmentUtil {
             try {
                 final Path fragmentPath = Path.of(pathString);
                 final Long modificationTime = fragmentPath.toRealPath().toFile().lastModified();
-                final String replacement = Files.readString(fragmentPath, StandardCharsets.UTF_8);
                 fragmentToModificationTime.put(fragmentPath, modificationTime);
+
+                final String replacement;
+                if (contentIsZippedAndBase64Encoded) {
+                    log.info("Config fragment {} is expected to be zipped and base64 encoded, extracting content.", pathString);
+                    replacement =
+                            deflate(Base64.decode(Files.readString(fragmentPath, StandardCharsets.UTF_8)))
+                                    .map(deflated -> new String(deflated, StandardCharsets.UTF_8))
+                                    .orElseThrow(UnrecoverableException::new);
+                } else {
+                    replacement = Files.readString(fragmentPath, StandardCharsets.UTF_8);
+                }
+
                 //sets replacement for this match
                 matcher.appendReplacement(resultString, escapeReplacement(replacement));
             } catch (IOException e) {
@@ -100,4 +119,16 @@ public FragmentResult(
             return renderResult;
         }
     }
+
+    public static Optional<byte[]> deflate(final byte[] zipData) {
+        try (final var zipStream = new ZipInputStream(new ByteArrayInputStream(zipData))) {
+            if (zipStream.getNextEntry() != null) {
+                return Optional.of(zipStream.readAllBytes());
+            }
+            return Optional.empty();
+        } catch (final IOException e) {
+            log.error("Error while extracting from ZIP: {}", e.getMessage());
+            return Optional.empty();
+        }
+    }
 }

hivemq-edge/src/test/java/com/hivemq/configuration/reader/AbstractConfigurationTest.java

@@ -88,6 +88,7 @@ public void setUp() throws Exception {
 
         final ConfigurationFile configurationFile = new ConfigurationFile(xmlFile);
         reader = new ConfigFileReaderWriter(
+                systemInformation,
                 configurationFile,
                 List.of(
                         new RestrictionConfigurator(restrictionsConfigurationService),

hivemq-edge/src/test/java/com/hivemq/configuration/reader/ConfigFileReaderTest.java

@@ -19,6 +19,7 @@
 import com.hivemq.configuration.entity.HiveMQConfigEntity;
 import com.hivemq.configuration.entity.adapter.MqttUserPropertyEntity;
 import com.hivemq.configuration.entity.adapter.ProtocolAdapterEntity;
+import com.hivemq.configuration.info.SystemInformation;
 import com.hivemq.exceptions.UnrecoverableException;
 import org.apache.commons.io.FileUtils;
 import org.jetbrains.annotations.NotNull;
@@ -145,7 +146,7 @@ public void whenUserPropertie_thenMapCorrectlyFilled() throws Exception {
         assertTrue(userPropertiesAfterReload.contains(new MqttUserPropertyEntity("my-name", "my-value2")));
     }
 
-    private static @NotNull ConfigFileReaderWriter getConfigFileReaderWriter(File tempFile) {
+    private static @NotNull ConfigFileReaderWriter getConfigFileReaderWriter(final File tempFile) {
         final ConfigurationFile configurationFile = new ConfigurationFile(tempFile);
 
         final RestrictionConfigurator restrictionConfigurator = mock(RestrictionConfigurator.class);
@@ -184,10 +185,12 @@ public void whenUserPropertie_thenMapCorrectlyFilled() throws Exception {
         final InternalConfigurator internalConfigurator = mock(InternalConfigurator.class);
         when(internalConfigurator.applyConfig(any())).thenReturn(Configurator.ConfigResult.SUCCESS);
 
-
-        final ConfigFileReaderWriter configFileReader = new ConfigFileReaderWriter(
+        final var sysInfo = mock(SystemInformation.class);
+        //ALways set to true for the test to ensure the fragment zipping code doesn't interfer with regular file rendering
+        when(sysInfo.isConfigFragmentBase64Zip()).thenReturn(true);
+        return new ConfigFileReaderWriter(
+                sysInfo,
                 configurationFile,
                 List.of());
-        return configFileReader;
     }
 }