refresh :: add ServiceProviderRefreshablePlace with "light" and "hard" refresh logic (#1073)

Author: dev-mlb

src/main/config/emissary.admin.ClassNameInventory.cfg

@@ -14,6 +14,7 @@ JavascriptEscapePlace = "emissary.transform.JavascriptEscapePlace"
 JsonEscapePlace       = "emissary.transform.JsonEscapePlace"
 KffHashPlace          = "emissary.place.KffHashPlace"
 PythonPlace           = "emissary.place.PythonPlace"
+RefreshablePlace      = "emissary.place.sample.RefreshablePlace"
 ToLowerPlace          = "emissary.place.sample.ToLowerPlace"
 ToUpperPlace          = "emissary.place.sample.ToUpperPlace"
 UnixCommandPlace      = "emissary.place.UnixCommandPlace"

src/main/config/emissary.place.sample.RefreshablePlace.cfg

@@ -0,0 +1,12 @@
+PLACE_NAME = "RefreshablePlace"
+SERVICE_NAME = "REFRESH_EXAMPLE"
+SERVICE_TYPE = "ANALYZE"
+SERVICE_DESCRIPTION = "refresh config stuff"
+SERVICE_COST = 50
+SERVICE_QUALITY = 60
+
+# Add Wildcard entry
+SERVICE_PROXY = "*"
+SERVICE_PROXY_DENY = "MYFORM"
+
+SAMPLE_CONFIG = 200

src/main/config/places.cfg

@@ -4,3 +4,5 @@ PLACE = "@{URL}/DropOffPlace"
 PLACE = "@{URL}/UnixFilePlace"
 PLACE = "@{URL}/JsonEscapePlace"
 PLACE = "@{URL}/VersionPlace"
+
+PLACE = "@{URL}/RefreshablePlace"

src/main/java/emissary/command/ServerCommand.java

@@ -16,6 +16,7 @@
 import java.util.LinkedHashSet;
 import java.util.Locale;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import static emissary.directory.EmissaryNode.STRICT_STARTUP_MODE;
 
@@ -37,6 +38,9 @@ public class ServerCommand extends ServiceCommand {
     @Option(names = {"-a", "--agents"}, description = "number of mobile agents (default is based on memory)\nDefault: ${DEFAULT-VALUE}")
     private int agents;
 
+    @Option(names = {"-t", "--timeout"}, description = "max amount of time to attempt a server refresh (in minutes) \nDefault: ${DEFAULT-VALUE}")
+    private int timeout;
+
     @Option(names = {"--dumpJettyBeans"}, description = "dump all the jetty beans that loaded\nDefault: ${DEFAULT-VALUE}")
     private boolean dumpJettyBeans = false;
 
@@ -65,6 +69,10 @@ public int getAgents() {
         return agents;
     }
 
+    public long getTimeout() {
+        return timeout;
+    }
+
     public boolean shouldDumpJettyBeans() {
         return dumpJettyBeans;
     }
@@ -81,6 +89,10 @@ public boolean shouldStrictMode() {
     @Override
     public void setupCommand() {
         setupHttp();
+        if (getTimeout() > 0) {
+            System.setProperty(EmissaryNode.NODE_REFRESH_TIMEOUT_PROPERTY, String.valueOf(TimeUnit.MINUTES.toMillis(getTimeout())));
+        }
+
         reinitLogback();
         setupServer();
     }
@@ -116,6 +128,16 @@ protected void startService() {
         EmissaryServer.init(this).startServer();
     }
 
+    @Override
+    protected void refreshService() {
+        EmissaryResponse response = performPost(getServiceRefreshEndpoint());
+        if (response.getStatus() != 200) {
+            LOG.error("Failed to {} Emissary services: {}", isInvalidate() ? "invalidate" : "refresh", response.getContentString());
+        } else {
+            LOG.info("{} Emissary services", isInvalidate() ? "Invalidating" : "Refreshing");
+        }
+    }
+
     @Override
     protected void pauseService() {
         setServerState(Pause.PAUSE);

src/main/java/emissary/command/ServiceCommand.java

@@ -11,6 +11,8 @@
 import picocli.CommandLine.Option;
 
 import static emissary.server.api.HealthCheckAction.HEALTH;
+import static emissary.server.api.Refresh.INVALIDATE;
+import static emissary.server.api.Refresh.REFRESH;
 import static emissary.server.api.Shutdown.SHUTDOWN;
 
 /**
@@ -24,13 +26,21 @@ public abstract class ServiceCommand extends HttpCommand {
     public static final String SERVICE_HEALTH_ENDPOINT = "/api/" + HEALTH;
     public static final String SERVICE_SHUTDOWN_ENDPOINT = "/api/" + SHUTDOWN;
     public static final String SERVICE_KILL_ENDPOINT = SERVICE_SHUTDOWN_ENDPOINT + "/force";
+    public static final String SERVICE_INVALIDATE_ENDPOINT = "/api/" + INVALIDATE;
+    public static final String SERVICE_REFRESH_ENDPOINT = "/api/" + REFRESH;
 
     @Option(names = {"--csrf"}, description = "disable csrf protection\nDefault: ${DEFAULT-VALUE}", arity = "1")
     private boolean csrf = true;
 
     @Option(names = {"--stop"}, description = "Shutdown the service\nDefault: ${DEFAULT-VALUE}")
     private boolean stop = false;
 
+    @Option(names = {"--invalidate"}, description = "Invalidate services that are refreshable\nDefault: ${DEFAULT-VALUE}")
+    private boolean invalidate = false;
+
+    @Option(names = {"--refresh"}, description = "Force refresh of services\nDefault: ${DEFAULT-VALUE}")
+    private boolean refresh = false;
+
     @Option(names = {"--kill"}, description = "Force the shutdown of the service\nDefault: ${DEFAULT-VALUE}")
     private boolean kill = false;
 
@@ -48,6 +58,14 @@ public boolean isStop() {
         return stop;
     }
 
+    public boolean isInvalidate() {
+        return invalidate;
+    }
+
+    public boolean isRefresh() {
+        return invalidate || refresh;
+    }
+
     public boolean isKill() {
         return kill;
     }
@@ -68,6 +86,10 @@ public String getServiceShutdownEndpoint() {
         return isKill() ? SERVICE_KILL_ENDPOINT : SERVICE_SHUTDOWN_ENDPOINT;
     }
 
+    public String getServiceRefreshEndpoint() {
+        return refresh ? SERVICE_REFRESH_ENDPOINT : SERVICE_INVALIDATE_ENDPOINT;
+    }
+
     public String getServiceName() {
         return getCommandName();
     }
@@ -109,6 +131,8 @@ public void run(CommandLine c) {
                     pauseService();
                 } else if (isUnpause()) {
                     unpauseService();
+                } else if (isRefresh()) {
+                    refreshService();
                 } else {
                     throw new EmissaryRuntimeException("Emissary " + getServiceName() + " is already running");
                 }
@@ -151,4 +175,11 @@ protected void unpauseService() {
         throw new UnsupportedOperationException("Unpause not implemented for " + getServiceName());
     }
 
+    /**
+     * A method that refreshes services
+     */
+    protected void refreshService() {
+        throw new UnsupportedOperationException("Refresh not implemented for " + getServiceName());
+    }
+
 }

src/main/java/emissary/directory/DirectoryEntry.java

@@ -528,6 +528,13 @@ public boolean isLocal() {
         return this.localPlace != null;
     }
 
+    /**
+     * Force a lookup from the namespace
+     */
+    public void clearLocalPlace() {
+        this.lookupAttempted = false;
+    }
+
     /**
      * Change the key such that the place specified by proxyKey acts as a proxy for the current key. We keep the same data
      * type, service type, service name and expense but change the place to the proxy

src/main/java/emissary/directory/DirectoryPlace.java

@@ -6,6 +6,7 @@
 import emissary.core.Namespace;
 import emissary.log.MDCConstants;
 import emissary.place.ServiceProviderPlace;
+import emissary.place.ServiceProviderRefreshablePlace;
 import emissary.server.mvc.adapters.DirectoryAdapter;
 
 import jakarta.annotation.Nullable;
@@ -979,6 +980,11 @@ protected List<DirectoryEntry> nextKeys(final String dataId, final IBaseDataObje
         // remove denied entries
         currentList.removeIf(de -> de.getLocalPlace() != null && de.getLocalPlace().isDenied(payload.currentForm()));
 
+        // test for invalidated places
+        currentList.stream().filter(e -> e.getLocalPlace() instanceof ServiceProviderRefreshablePlace
+                && ((ServiceProviderRefreshablePlace) e.getLocalPlace()).isInvalidated())
+                .forEach(this::handleEntryRefresh);
+
         if (currentList.isEmpty()) {
             logger.debug("nextKeys - no non-DENIED entries found here for {}", dataId);
             return List.of();
@@ -1036,6 +1042,19 @@ protected List<DirectoryEntry> nextKeys(final String dataId, final IBaseDataObje
         return keyList;
     }
 
+    protected synchronized void handleEntryRefresh(final DirectoryEntry entry) {
+        // attempt to see if the place has been refreshed already
+        entry.clearLocalPlace();
+
+        final ServiceProviderRefreshablePlace place = (ServiceProviderRefreshablePlace) entry.getLocalPlace();
+        if (place.isInvalidated()) {
+            logger.debug("{} has been invalidated, attempting refresh ....", entry);
+            place.refresh();
+            entry.clearLocalPlace();
+            logger.debug("{} has been refreshed", entry);
+        }
+    }
+
     /**
      * Get the possibly wildcarded DirectoryEntryList for the dataId
      *
@@ -1186,10 +1205,6 @@ public int removePlaces(final List<String> keys) {
      */
     @Override
     public int irdRemovePlaces(@Nullable final List<String> keys, final boolean propagating) {
-        if (this.emissaryNode.isStandalone()) {
-            logger.debug("Cannot remove remote places in standalone nodes");
-            return 0;
-        }
 
         if ((keys == null) || keys.isEmpty()) {
             logger.warn("Ignoring null or empty key list for irdRemovePlaces");

src/main/java/emissary/directory/EmissaryNode.java

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Hold some details about being a P2P node in the emissary network The order of preference to find the node
@@ -52,6 +53,9 @@ public class EmissaryNode {
     /** Node service type property */
     public static final String NODE_SERVICE_TYPE_PROPERTY = "emissary.node.service.type";
 
+    /** Node refresh timeout property */
+    public static final String NODE_REFRESH_TIMEOUT_PROPERTY = "emissary.node.refresh.timeout";
+
     /** Node service type is {@value} */
     public static final String DEFAULT_NODE_SERVICE_TYPE = "server";
 
@@ -64,6 +68,9 @@ public class EmissaryNode {
     /** Property that determines if server will shut down in the event a place fails to start */
     public static final String STRICT_STARTUP_MODE = "strict.mode";
 
+    /** Property that sets the max amount of time to wait for a refresh before a failure condition */
+    public static final long DEFAULT_REFRESH_TIMEOUT = TimeUnit.MINUTES.toMillis(30);
+
     public enum Mode {
         STANDALONE, CLUSTER;
     }
@@ -80,7 +87,7 @@ public enum Mode {
     protected boolean nodeNameIsDefault = false;
     @Nullable
     protected String nodeServiceType = null;
-
+    protected long nodeRefreshTimeout;
     protected boolean strictStartupMode = false;
 
     public EmissaryNode() {
@@ -109,6 +116,7 @@ public EmissaryNode(Mode nodeMode) {
         this.nodePort = Integer.getInteger(NODE_PORT_PROPERTY, -1).intValue();
         this.nodeType = System.getProperty("os.name", DEFAULT_NODE_TYPE).toLowerCase(Locale.getDefault()).replace(' ', '_');
         this.nodeServiceType = System.getProperty(NODE_SERVICE_TYPE_PROPERTY, DEFAULT_NODE_SERVICE_TYPE);
+        this.nodeRefreshTimeout = Long.getLong(NODE_REFRESH_TIMEOUT_PROPERTY, DEFAULT_REFRESH_TIMEOUT);
         this.strictStartupMode = Boolean.parseBoolean(System.getProperty(STRICT_STARTUP_MODE, String.valueOf(false)));
     }
 
@@ -144,6 +152,13 @@ public String getNodeScheme() {
         return this.nodeScheme;
     }
 
+    /**
+     * The node max amount of time to wait for a refresh before throwing a failure condition
+     */
+    public long getNodeRefreshTimeout() {
+        return this.nodeRefreshTimeout;
+    }
+
     /**
      * Get the value as a url
      */