US1065453: Support BE preprocessor rules (#3)

- Adds support for new preprocessor rule objects
- Updates logging from log4j to JUL
- Bump java-lib to 2026-2.1

Author: joannatk

proxy/pom.xml

@@ -4,7 +4,7 @@
 
   <groupId>com.wavefront</groupId>
   <artifactId>proxy</artifactId>
-  <version>13.8-RC1-SNAPSHOT</version>
+  <version>14.0-SNAPSHOT</version>
 
   <name>Wavefront Proxy</name>
   <description>Service for batching and relaying metric traffic to Wavefront</description>
@@ -452,7 +452,7 @@
     <dependency>
       <groupId>com.wavefront</groupId>
       <artifactId>java-lib</artifactId>
-      <version>2023-08.2</version>
+      <version>2026-2.1</version>
     </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.module</groupId>

proxy/src/main/java/com/wavefront/agent/AbstractAgent.java

@@ -20,10 +20,10 @@
 import com.wavefront.agent.data.EntityPropertiesFactoryImpl;
 import com.wavefront.agent.logsharvesting.InteractiveLogsTester;
 import com.wavefront.agent.preprocessor.InteractivePreprocessorTester;
-import com.wavefront.agent.preprocessor.LineBasedAllowFilter;
-import com.wavefront.agent.preprocessor.LineBasedBlockFilter;
-import com.wavefront.agent.preprocessor.PreprocessorConfigManager;
-import com.wavefront.agent.preprocessor.PreprocessorRuleMetrics;
+import com.wavefront.api.agent.preprocessor.LineBasedAllowFilter;
+import com.wavefront.api.agent.preprocessor.LineBasedBlockFilter;
+import com.wavefront.agent.preprocessor.ProxyPreprocessorConfigManager;
+import com.wavefront.api.agent.preprocessor.PreprocessorRuleMetrics;
 import com.wavefront.agent.queueing.QueueExporter;
 import com.wavefront.agent.queueing.SQSQueueFactoryImpl;
 import com.wavefront.agent.queueing.TaskQueueFactory;
@@ -71,7 +71,8 @@ public abstract class AbstractAgent {
   protected APIContainer apiContainer;
   protected final List<ExecutorService> managedExecutors = new ArrayList<>();
   protected final List<Runnable> shutdownTasks = new ArrayList<>();
-  protected final PreprocessorConfigManager preprocessors = new PreprocessorConfigManager();
+  protected final ProxyPreprocessorConfigManager preprocessors =
+          new ProxyPreprocessorConfigManager();
   protected final ValidationConfiguration validationConfiguration = new ValidationConfiguration();
   protected final Map<String, EntityPropertiesFactory> entityPropertiesFactoryMap =
       Maps.newHashMap();
@@ -142,15 +143,18 @@ void initSslContext() throws SSLException {
 
   private void initPreprocessors() {
     String configFileName = proxyConfig.getPreprocessorConfigFile();
-    if (configFileName != null) {
+    if (ProxyCheckInScheduler.isRulesSetInFE.get()) {
+      logger.info("Preprocessor configuration was set in FE. Skipping reading from file " + configFileName);
+    } else if (configFileName != null) {
       try {
         preprocessors.loadFile(configFileName);
         preprocessors.setUpConfigFileMonitoring(configFileName, 5000); // check every 5s
       } catch (FileNotFoundException ex) {
         throw new RuntimeException(
             "Unable to load preprocessor rules - file does not exist: " + configFileName);
       }
-      logger.info("Preprocessor configuration loaded from " + configFileName);
+      logger.log(Level.INFO, () -> String.format("Preprocessor configuration loaded from %s",
+              (ProxyCheckInScheduler.isRulesSetInFE.get()) ?  "FE rule" : configFileName));
     }
 
     // convert block/allow list fields to filters for full backwards compatibility.

proxy/src/main/java/com/wavefront/agent/ProxyCheckInScheduler.java

@@ -4,12 +4,13 @@
 import static org.apache.commons.lang3.ObjectUtils.firstNonNull;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Strings;
 import com.google.common.base.Throwables;
 import com.google.common.collect.Maps;
 import com.wavefront.agent.api.APIContainer;
-import com.wavefront.agent.preprocessor.PreprocessorConfigManager;
+import com.wavefront.agent.preprocessor.ProxyPreprocessorConfigManager;
 import com.wavefront.api.agent.AgentConfiguration;
 import com.wavefront.api.agent.ValidationConfiguration;
 import com.wavefront.common.Clock;
@@ -32,8 +33,8 @@
 import javax.ws.rs.ClientErrorException;
 import javax.ws.rs.ProcessingException;
 import org.apache.commons.lang.StringUtils;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 /**
  * Registers the proxy with the back-end, sets up regular "check-ins" (every minute), transmits
@@ -42,7 +43,7 @@
  * @author vasily@wavefront.com
  */
 public class ProxyCheckInScheduler {
-  private static final Logger logger = LogManager.getLogger("proxy");
+  private static final Logger logger = Logger.getLogger("proxy");
   private static final int MAX_CHECKIN_ATTEMPTS = 5;
 
   /**
@@ -68,7 +69,10 @@ public class ProxyCheckInScheduler {
   private volatile JsonNode agentMetrics;
   private boolean retryImmediately = false;
 
+  // check if preprocessor rules need to be sent to update BE
   public static AtomicBoolean preprocessorRulesNeedUpdate = new AtomicBoolean(false);
+  // check if rules are set from FE/API
+  public static AtomicBoolean isRulesSetInFE = new AtomicBoolean(false);
 
   /**
    * @param proxyId Proxy UUID.
@@ -134,19 +138,60 @@ public void shutdown() {
     executor.shutdown();
   }
 
+  /**
+   * Initial sending of preprocessor rules. Will always check local location for preprocessor rule.
+   */
+  public void sendPreprocessorRules() {
+    if (preprocessorRulesNeedUpdate.getAndSet(false)) {
+      try {
+        JsonNode rulesNode = createRulesNode(ProxyPreprocessorConfigManager.getProxyConfigRules(), null);
+        apiContainer
+                .getProxyV2APIForTenant(APIContainer.CENTRAL_TENANT_NAME)
+                .proxySavePreprocessorRules(
+                        proxyId,
+                        rulesNode
+                );
+      } catch (javax.ws.rs.NotFoundException ex) {
+        logger.warning("'proxySavePreprocessorRules' api end point not found");
+      }
+    }
+  }
+
   /** Send preprocessor rules */
-  private void sendPreprocessorRules() {
+  private void sendPreprocessorRules(AgentConfiguration agentConfiguration) {
     if (preprocessorRulesNeedUpdate.getAndSet(false)) {
+      String preprocessorRules = null;
+      if (agentConfiguration.getPreprocessorRules() != null) {
+        // reading rules from BE if sent from BE
+        preprocessorRules = agentConfiguration.getPreprocessorRules();
+      } else {
+        // reading local file's rule
+        preprocessorRules = ProxyPreprocessorConfigManager.getProxyConfigRules();
+      }
       try {
+        JsonNode rulesNode = createRulesNode(preprocessorRules, agentConfiguration.getPreprocessorRulesId());
         apiContainer
-            .getProxyV2APIForTenant(APIContainer.CENTRAL_TENANT_NAME)
-            .proxySavePreprocessorRules(proxyId, PreprocessorConfigManager.getJsonRules());
+                .getProxyV2APIForTenant(APIContainer.CENTRAL_TENANT_NAME)
+                .proxySavePreprocessorRules(
+                        proxyId,
+                        rulesNode
+                );
       } catch (javax.ws.rs.NotFoundException ex) {
-        logger.debug("'proxySavePreprocessorRules' api end point not found", ex);
+        logger.warning("'proxySavePreprocessorRules' api end point not found");
       }
     }
   }
 
+  private JsonNode createRulesNode(String preprocessorRules, String proxyId) {
+    Map<String, String> fieldsMap = new HashMap<>();
+    fieldsMap.put("proxyRules", preprocessorRules);
+    if (proxyConfig.getPreprocessorConfigFile() != null) fieldsMap.put("proxyRulesFilePath", proxyConfig.getPreprocessorConfigFile());
+    if (proxyId != null) fieldsMap.put("proxyRulesId", proxyId);
+
+    ObjectMapper objectMapper = new ObjectMapper();
+    return objectMapper.valueToTree(fieldsMap);
+  }
+
   /**
    * Perform agent check-in and fetch configuration of the daemon from remote server.
    *
@@ -335,15 +380,15 @@ private Map<String, AgentConfiguration> checkin() {
         if (StringUtils.isBlank(logServerIngestionURL)
             || StringUtils.isBlank(logServerIngestionToken)) {
           proxyConfig.setReceivedLogServerDetails(false);
-          logger.error(
+          logger.severe(
               WARNING_MSG
                   + " To ingest logs to the log server, please provide "
                   + "logServerIngestionToken & logServerIngestionURL in the proxy configuration.");
         }
       }
     } else if (StringUtils.isBlank(logServerIngestionURL)
         || StringUtils.isBlank(logServerIngestionToken)) {
-      logger.warn(
+      logger.severe(
           WARNING_MSG
               + " Proxy will not be ingesting data to the log server as it did "
               + "not receive at least one of the values during check-in.");
@@ -358,7 +403,6 @@ private Map<String, AgentConfiguration> checkin() {
   void updateConfiguration() {
     try {
       Map<String, AgentConfiguration> configList = checkin();
-      sendPreprocessorRules();
       if (configList != null && !configList.isEmpty()) {
         AgentConfiguration config;
         for (Map.Entry<String, AgentConfiguration> configEntry : configList.entrySet()) {
@@ -368,27 +412,53 @@ void updateConfiguration() {
             continue;
           }
           if (configEntry.getKey().equals(APIContainer.CENTRAL_TENANT_NAME)) {
-            if (logger.isDebugEnabled()) {
-              logger.debug("Server configuration getShutOffAgents: " + config.getShutOffAgents());
-              logger.debug("Server configuration isTruncateQueue: " + config.isTruncateQueue());
+            if (logger.isLoggable(Level.FINE)) {
+              logger.fine("Server configuration getShutOffAgents: " + config.getShutOffAgents());
+              logger.fine("Server configuration isTruncateQueue: " + config.isTruncateQueue());
             }
             if (config.getShutOffAgents()) {
-              logger.warn(
+              logger.severe(
                   firstNonNull(
                       config.getShutOffMessage(),
                       "Shutting down: Server side flag indicating proxy has to shut down."));
               shutdownHook.run();
             } else if (config.isTruncateQueue()) {
-              logger.warn(
+              logger.severe(
                   "Truncating queue: Server side flag indicating proxy queue has to be truncated.");
               truncateBacklog.run();
             }
           }
           agentConfigurationConsumer.accept(configEntry.getKey(), config);
+
+          // Check if preprocessor rules were set on server side
+          String checkPreprocessorRules = config.getPreprocessorRules();
+          if (checkPreprocessorRules != null && !checkPreprocessorRules.isEmpty()) {
+            AgentConfiguration finalConfig = config;
+            logger.log(Level.INFO, () -> String.format("New preprocessor rules detected during checkin. Setting new preprocessor rule %s",
+                    (finalConfig.getPreprocessorRulesId() != null && !finalConfig.getPreprocessorRulesId().isEmpty()) ? finalConfig.getPreprocessorRulesId() : ""));
+            // future implementation, can send timestamp through AgentConfig and skip reloading if rule unchanged
+            isRulesSetInFE.set(true);
+            // indicates will need to sendPreprocessorRules()
+            preprocessorRulesNeedUpdate.set(true);
+          } else {
+            // was previously reading from BE
+            if (isRulesSetInFE.get()) {
+              if (proxyConfig.getPreprocessorConfigFile() == null || proxyConfig.getPreprocessorConfigFile().isEmpty()) {
+                logger.info("No preprocessor rules detected during checkin, and no rules file found.");
+              } else {
+                logger.log(Level.INFO, () -> String.format("Reverting back to reading rules from file %s", proxyConfig.getPreprocessorConfigFile()));
+              }
+              // indicates that previously read from BE, now switching back to reading from file.
+              isRulesSetInFE.set(false);
+              preprocessorRulesNeedUpdate.set(true);
+            }
+          }
+          // will always send to BE in order to update Agent with latest rule
+          sendPreprocessorRules(config);
         }
       }
     } catch (Exception e) {
-      logger.error("Exception occurred during configuration update", e);
+      logger.log(Level.SEVERE, "Exception occurred during configuration update", e);
     }
   }
 
@@ -405,13 +475,13 @@ void updateProxyMetrics() {
         retries.set(0);
       }
     } catch (Exception ex) {
-      logger.error("Could not generate proxy metrics", ex);
+      logger.log(Level.SEVERE, "Could not generate proxy metrics", ex);
     }
   }
 
   private void checkinError(String errMsg) {
-    if (successfulCheckIns.get() == 0) logger.error(Strings.repeat("*", errMsg.length()));
-    logger.error(errMsg);
-    if (successfulCheckIns.get() == 0) logger.error(Strings.repeat("*", errMsg.length()));
+    if (successfulCheckIns.get() == 0) logger.severe(Strings.repeat("*", errMsg.length()));
+    logger.severe(errMsg);
+    if (successfulCheckIns.get() == 0) logger.severe(Strings.repeat("*", errMsg.length()));
   }
 }

proxy/src/main/java/com/wavefront/agent/ProxySendConfigScheduler.java

@@ -5,13 +5,12 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import org.apache.logging.log4j.Level;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 public class ProxySendConfigScheduler {
   private static final Logger logger =
-      LogManager.getLogger(ProxySendConfigScheduler.class.getCanonicalName());
+      Logger.getLogger(ProxySendConfigScheduler.class.getCanonicalName());
   private boolean successful = false;
   private final ScheduledExecutorService executor;
   private final Runnable task;
@@ -28,13 +27,13 @@ public ProxySendConfigScheduler(
             successful = true;
             logger.info("Configuration sent to the server successfully.");
           } catch (javax.ws.rs.NotFoundException ex) {
-            logger.debug("'proxySaveConfig' api end point not found", ex);
+            logger.log(Level.FINE, "'proxySaveConfig' api end point not found", ex);
             successful = true;
           } catch (Throwable e) {
-            logger.warn(
+            logger.severe(
                 "Can't send the Proxy configuration to the server, retrying in 60 seconds. "
                     + e.getMessage());
-            logger.log(Level.DEBUG, "Exception: ", e);
+            logger.log(Level.FINE, "Exception: ", e);
           }
 
           if (successful) {

proxy/src/main/java/com/wavefront/agent/PushAgent.java

@@ -41,10 +41,10 @@
 import com.wavefront.agent.listeners.tracing.*;
 import com.wavefront.agent.logsharvesting.FilebeatIngester;
 import com.wavefront.agent.logsharvesting.LogsIngester;
-import com.wavefront.agent.preprocessor.PreprocessorRuleMetrics;
-import com.wavefront.agent.preprocessor.ReportPointAddPrefixTransformer;
-import com.wavefront.agent.preprocessor.ReportPointTimestampInRangeFilter;
-import com.wavefront.agent.preprocessor.SpanSanitizeTransformer;
+import com.wavefront.api.agent.preprocessor.PreprocessorRuleMetrics;
+import com.wavefront.api.agent.preprocessor.ReportPointAddPrefixTransformer;
+import com.wavefront.api.agent.preprocessor.ReportPointTimestampInRangeFilter;
+import com.wavefront.api.agent.preprocessor.SpanSanitizeTransformer;
 import com.wavefront.agent.queueing.*;
 import com.wavefront.agent.sampler.SpanSampler;
 import com.wavefront.agent.sampler.SpanSamplerUtils;
@@ -71,6 +71,7 @@
 import io.netty.handler.codec.http.cors.CorsConfigBuilder;
 import io.netty.handler.ssl.SslContext;
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.net.BindException;
 import java.net.InetAddress;
 import java.nio.ByteOrder;
@@ -79,6 +80,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.logging.Level;
@@ -157,6 +159,7 @@ public class PushAgent extends AbstractAgent {
   private Logger blockedHistogramsLogger;
   private Logger blockedSpansLogger;
   private Logger blockedLogsLogger;
+  private AtomicBoolean usingLocalFileRules = new AtomicBoolean(true);
 
   public static void main(String[] args) {
     // Start the ssh daemon
@@ -1881,6 +1884,18 @@ private void registerPrefixFilter(String strPort) {
   @Override
   protected void processConfiguration(String tenantName, AgentConfiguration config) {
     try {
+      boolean configHasPreprocessorRules = config.getPreprocessorRules() != null && !config.getPreprocessorRules().isEmpty();
+//       apply new preprocessor rules after checkin
+      if (ProxyCheckInScheduler.isRulesSetInFE.get() && configHasPreprocessorRules) {
+        loadPreprocessors(config.getPreprocessorRules(), true);
+      }
+      // check if we want to read local file
+      else if (!ProxyCheckInScheduler.isRulesSetInFE.get() &&
+              proxyConfig.getPreprocessorConfigFile() != null &&
+              !usingLocalFileRules.get()) { // are we already reading local file
+        loadPreprocessors(proxyConfig.getPreprocessorConfigFile(), false);
+      }
+
       Long pointsPerBatch = config.getPointsPerBatch();
       EntityPropertiesFactory tenantSpecificEntityProps =
           entityPropertiesFactoryMap.get(tenantName);
@@ -2140,4 +2155,20 @@ protected void stopListener(int port) {
   protected void truncateBacklog() {
     senderTaskFactory.truncateBuffers();
   }
+
+  private void loadPreprocessors(String preprocessorStr, boolean readfromFE) {
+    try {
+      if (readfromFE && preprocessorStr != null) {
+        // preprocessorStr is str of rules itself
+        preprocessors.loadFERules(preprocessorStr);
+        usingLocalFileRules.set(false);
+      } else {
+        // preprocessorStr is file path
+        preprocessors.loadFile(preprocessorStr);
+        usingLocalFileRules.set(true);
+      }
+    } catch (FileNotFoundException e) {
+      throw new RuntimeException("Unable to load preprocessor rules - file does not exist: " + preprocessorStr);
+    }
+  }
 }