Detect frozen PVs with an independent CA context

README.md

@@ -69,6 +69,11 @@ A client that stops reading makes writes to it block once the network buffers fi
 
 By default the response is 200 whenever the server is up, which suits load balancers: an IOC being down affects every instance alike.  With `?strict=true` the response is 503 when a PV that was connected has disconnected, which suits monitoring that should alert on that, such as Nagios.  PVs that never connected are listed but don't fail strict mode, since they may simply not exist.
 
+#### Frozen PVs
+A PV is frozen when its monitor has stopped working while the IOC still serves it, so restarting epics2web would be expected to fix it.  epics2web looks for them every **FROZEN_CHECK_SECONDS** (default 10) by giving suspicious PVs a short-lived subscription in a second, independent CA context with its own connections: PVs disconnected for longer than the grace period, and PVs that used to update but have been quiet for 6 check intervals.  A PV is frozen if its monitor stays disconnected while the independent subscription connects, or if the independent subscription receives a change the monitor doesn't.  An IOC that's down doesn't make its PVs frozen.  Through a gateway, both contexts reach the PV through the gateway, so this finds problems between epics2web and the gateway, not inside it.
+
+Frozen PVs are logged as warnings and listed by `/healthcheck` with `frozen`, `frozen_minutes` and `frozen_reason`.  They don't change the default or strict response; with `?frozen=true` the response is 503 when a PV is frozen, for automation that restarts epics2web.  **FROZEN_PV_CHECK**=false turns detection off.
+
 ### Logging
 This app is designed to run on Tomcat so [Tomcat logging configuration](https://tomcat.apache.org/tomcat-9.0-doc/logging.html) applies.  We use the built-in JVM logging library, which Tomcat uses with some slight modifications to support separate classloaders.  In the past we bundled an application [logging.properites](https://github.com/JeffersonLab/epics2web/blob/956894699ef1b303907a04720aeb50260ffa72b1/src/main/resources/logging.properties) inside the epics2web.war file.  We no longer do that because it then appears to require repackaging/rebuilding a new version of the app to modify the logging config as the app bundled config overrides the global Tomcat config at conf/logging.properties.  The recommend logging strategy is to now make configuration in the global Tomcat config so as to make it easy to modify logging levels.  An app specific handler can be created.  The global configuration location is generally set by the Tomcat default start script via JVM system properties.  The system properties should look something like: 
 - `-Djava.util.logging.config.file=/usr/share/tomcat/conf/logging.properties`

build.yaml

@@ -15,6 +15,7 @@ services:
       WEBSOCKET_SEND_TIMEOUT_SECONDS: 3
       # Short, so HealthcheckTest runs in seconds
       HEALTHCHECK_GRACE_SECONDS: 2
+      FROZEN_CHECK_SECONDS: 1
     build:
       context: .
       dockerfile: Dockerfile

src/integration/java/org/jlab/epics2web/FrozenPvDetectionTest.java

@@ -0,0 +1,137 @@
+package org.jlab.epics2web;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import com.cosylab.epics.caj.CAJContext;
+import gov.aps.jca.JCALibrary;
+import gov.aps.jca.Monitor;
+import gov.aps.jca.configuration.DefaultConfiguration;
+import gov.aps.jca.dbr.DBR;
+import gov.aps.jca.dbr.DBRType;
+import java.lang.reflect.Field;
+import java.time.Clock;
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import org.jlab.epics2web.epics.CaProbeFactory;
+import org.jlab.epics2web.epics.ChannelManager;
+import org.jlab.epics2web.epics.ChannelMonitor;
+import org.jlab.epics2web.epics.FrozenPvDetector;
+import org.jlab.epics2web.epics.PvListener;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+
+/**
+ * Runs FrozenPvDetector in this JVM against the test IOC, through its published ports, with two
+ * real CA contexts. A lost subscription is simulated by clearing a monitor's CAJ subscription
+ * without telling the monitor, as a CA client bug might.
+ */
+public class FrozenPvDetectionTest {
+
+  private static final String MISSING = "epics2web:test:frozen:missing";
+
+  @Rule public Timeout globalTimeout = Timeout.seconds(90);
+
+  private CAJContext monitorContext;
+  private CAJContext probeContext;
+  private ScheduledExecutorService timeoutExecutor;
+  private ExecutorService callbackExecutor;
+  private ChannelManager manager;
+  private FrozenPvDetector detector;
+  private final PvListener listener = new NoopListener();
+
+  @BeforeClass
+  public static void disableRepeater() {
+    System.setProperty("CA_DISABLE_REPEATER", "true"); // As in the unit tests (#32)
+  }
+
+  @Before
+  public void setUp() throws Exception {
+    monitorContext = newContext();
+    probeContext = newContext();
+    timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
+    callbackExecutor = Executors.newCachedThreadPool();
+    manager = new ChannelManager(monitorContext, timeoutExecutor, callbackExecutor);
+    detector =
+        new FrozenPvDetector(
+            () -> FrozenPvDetector.viewsOf(manager.getMonitorMap()),
+            new CaProbeFactory(probeContext),
+            Clock.systemUTC(),
+            Duration.ofSeconds(2),
+            Duration.ofSeconds(1),
+            20);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    detector.close();
+    manager.removeAll(listener);
+    timeoutExecutor.shutdownNow();
+    callbackExecutor.shutdownNow();
+    monitorContext.destroy();
+    probeContext.destroy();
+  }
+
+  @Test
+  public void lostSubscriptionIsFrozenAndNothingElseIs() throws Exception {
+    manager.addPv(listener, "HELLO"); // Changes every 0.2 s
+    manager.addPv(listener, "channel1"); // Never changes
+    manager.addPv(listener, MISSING); // Never connects
+
+    // Working monitors, a PV that doesn't change, and a PV that doesn't exist aren't frozen. 10 s
+    // is longer than the 6 s after which a PV counts as quiet, and the 2 s grace period.
+    checkFor(10);
+    assertEquals(Map.of(), detector.getFrozen());
+    assertTrue(manager.getMonitorMap().get("HELLO").getUpdateCount() > 10);
+
+    loseSubscription(manager.getMonitorMap().get("HELLO"));
+
+    long deadline = System.currentTimeMillis() + 30_000;
+    while (!detector.getFrozen().containsKey("HELLO") && System.currentTimeMillis() < deadline) {
+      checkFor(1);
+    }
+    assertEquals(Set.of("HELLO"), detector.getFrozen().keySet());
+    assertTrue(
+        detector.getFrozen().get("HELLO").reason(),
+        detector.getFrozen().get("HELLO").reason().contains("didn't receive"));
+  }
+
+  private void checkFor(int seconds) throws InterruptedException {
+    for (int i = 0; i < seconds; i++) {
+      Thread.sleep(1_000);
+      detector.check();
+    }
+  }
+
+  /** Cancel the monitor's CAJ subscription without telling the monitor. */
+  private static void loseSubscription(ChannelMonitor monitor) throws Exception {
+    Field field = ChannelMonitor.class.getDeclaredField("monitor");
+    field.setAccessible(true);
+    ((Monitor) field.get(monitor)).clear();
+  }
+
+  private static CAJContext newContext() throws Exception {
+    DefaultConfiguration config = new DefaultConfiguration("test");
+    config.setAttribute("class", JCALibrary.CHANNEL_ACCESS_JAVA);
+    config.setAttribute("addr_list", "127.0.0.1");
+    config.setAttribute("auto_addr_list", "false");
+    return (CAJContext) JCALibrary.getInstance().createContext(config);
+  }
+
+  private static class NoopListener implements PvListener {
+    @Override
+    public void notifyPvInfo(
+        String pv, boolean couldConnect, DBRType type, Integer count, String[] enumLabels) {}
+
+    @Override
+    public void notifyPvUpdate(String pv, DBR dbr) {}
+  }
+}

src/integration/java/org/jlab/epics2web/HealthcheckTest.java

@@ -1,6 +1,7 @@
 package org.jlab.epics2web;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -77,6 +78,12 @@ public void disconnectedPvFailsStrictModeOnly() throws Exception {
       assertTrue("Reported after only " + reportedAfter + " ms", reportedAfter >= 2_000);
       assertEquals(200, get("healthcheck").statusCode());
       assertEquals(503, get("healthcheck?strict=true").statusCode());
+
+      // The IOC is down, so restarting epics2web wouldn't help: not frozen. With
+      // FROZEN_CHECK_SECONDS 1, the detector has probed it within 2 s of the grace period ending.
+      Thread.sleep(3_000);
+      assertFalse(entry(get("healthcheck"), "channel1").containsKey("frozen"));
+      assertEquals(200, get("healthcheck?frozen=true").statusCode());
     } finally {
       docker("start", "softioc");
       waitForEntry("channel1", false); // Reconnected
@@ -85,6 +92,25 @@ public void disconnectedPvFailsStrictModeOnly() throws Exception {
     assertEquals(200, get("healthcheck?strict=true").statusCode());
   }
 
+  /**
+   * Working monitors aren't frozen. HELLO changes every 0.2 s; channel1 never changes. 10 s is
+   * longer than the 6 s after which a quiet PV is probed, with FROZEN_CHECK_SECONDS 1.
+   */
+  @Test
+  public void workingMonitorsAreNotFrozen() throws Exception {
+    WebSocket hello = monitor("HELLO");
+    WebSocket channel1 = monitor("channel1");
+    try {
+      Thread.sleep(10_000);
+      assertEquals(200, get("healthcheck?frozen=true").statusCode());
+      assertNull(entry(get("healthcheck"), "HELLO"));
+      assertNull(entry(get("healthcheck"), "channel1"));
+    } finally {
+      hello.abort();
+      channel1.abort();
+    }
+  }
+
   private static WebSocket monitor(String pv) {
     WebSocket socket =
         HTTP.newWebSocketBuilder()

src/main/java/org/jlab/epics2web/Application.java

@@ -15,6 +15,7 @@
 import jakarta.websocket.SendResult;
 import jakarta.websocket.Session;
 import java.io.IOException;
+import java.time.Clock;
 import java.time.Duration;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
@@ -26,8 +27,10 @@
 import java.util.concurrent.locks.LockSupport;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import org.jlab.epics2web.epics.CaProbeFactory;
 import org.jlab.epics2web.epics.ChannelManager;
 import org.jlab.epics2web.epics.ContextFactory;
+import org.jlab.epics2web.epics.FrozenPvDetector;
 import org.jlab.epics2web.websocket.WebSocketSessionManager;
 import org.jlab.epics2web.websocket.WriteQueue;
 import org.jlab.epics2web.websocket.WriteStrategy;
@@ -46,6 +49,9 @@ public class Application implements ServletContextListener {
   public static ChannelManager channelManager = null;
   public static WebSocketSessionManager sessionManager = null;
 
+  /** Null if frozen PV detection is off. */
+  public static FrozenPvDetector frozenPvDetector = null;
+
   private static final int TIMEOUT_EXECUTOR_POOL_SIZE = 1;
   private static final Logger LOGGER = Logger.getLogger(Application.class.getName());
 
@@ -67,11 +73,23 @@ public class Application implements ServletContextListener {
   public static final long HEALTHCHECK_GRACE_SECONDS =
       getSecondsFromEnv("HEALTHCHECK_GRACE_SECONDS", 30);
 
+  /** Whether to look for frozen PVs; FROZEN_PV_CHECK=false turns it off. */
+  static final boolean FROZEN_PV_CHECK =
+      !"false".equalsIgnoreCase(System.getenv("FROZEN_PV_CHECK"));
+
+  /** How often to look for frozen PVs; FrozenPvDetector derives its other timings from it. */
+  static final long FROZEN_CHECK_SECONDS = getSecondsFromEnv("FROZEN_CHECK_SECONDS", 10);
+
+  /** The most PVs probed at once in the independent context. */
+  private static final int FROZEN_CHECK_MAX_PROBES = 20;
+
   private static ScheduledExecutorService timeoutExecutor = null;
   private static ExecutorService callbackExecutor = null;
   private static ExecutorService writerExecutor = null;
   private static ScheduledExecutorService sessionCheckExecutor = null;
   private static ExecutorService pingExecutor = null;
+  private static ScheduledExecutorService frozenCheckExecutor = null;
+  private static volatile CAJContext probeContext = null;
   private static ContextFactory factory = null;
   private static volatile CAJContext context = null;
 
@@ -205,6 +223,10 @@ public void contextInitialized(ServletContextEvent sce) {
       LOGGER.log(Level.SEVERE, "Unable to register context callbacks", e);
     }
 
+    if (FROZEN_PV_CHECK) {
+      startFrozenPvDetection();
+    }
+
     if (WRITE_STRATEGY == WriteStrategy.ASYNC_QUEUE) {
       writerExecutor.execute(
           new Runnable() {
@@ -296,6 +318,22 @@ public void contextDestroyed(ServletContextEvent sce) {
       pingExecutor.shutdownNow();
     }
 
+    if (frozenCheckExecutor != null) {
+      frozenCheckExecutor.shutdownNow();
+    }
+
+    if (frozenPvDetector != null) {
+      frozenPvDetector.close();
+    }
+
+    if (probeContext != null) {
+      try {
+        probeContext.destroy();
+      } catch (CAException e) {
+        LOGGER.log(Level.WARNING, "Unable to destroy probe context", e);
+      }
+    }
+
     if (timeoutExecutor != null) {
       try {
         if (!timeoutExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
@@ -329,6 +367,46 @@ public void contextDestroyed(ServletContextEvent sce) {
     }
   }
 
+  /**
+   * Look for frozen PVs with subscriptions in a second CA context, which has its own virtual
+   * circuits. See FrozenPvDetector.
+   */
+  private void startFrozenPvDetection() {
+    try {
+      probeContext = factory.newContext();
+    } catch (Exception e) {
+      LOGGER.log(Level.SEVERE, "Unable to create CA context for frozen PV detection", e);
+      return;
+    }
+
+    frozenPvDetector =
+        new FrozenPvDetector(
+            () -> FrozenPvDetector.viewsOf(channelManager.getMonitorMap()),
+            new CaProbeFactory(probeContext),
+            Clock.systemUTC(),
+            Duration.ofSeconds(HEALTHCHECK_GRACE_SECONDS),
+            Duration.ofSeconds(FROZEN_CHECK_SECONDS),
+            FROZEN_CHECK_MAX_PROBES);
+
+    // Its own thread: closing a probe can wait briefly for the IOC
+    frozenCheckExecutor =
+        Executors.newSingleThreadScheduledExecutor(
+            new CustomPrefixThreadFactory("Frozen-PV-Check-"));
+    frozenCheckExecutor.scheduleWithFixedDelay(
+        () -> {
+          try {
+            frozenPvDetector.check();
+          } catch (RuntimeException e) { // An exception would cancel the schedule
+            LOGGER.log(Level.WARNING, "Unable to check for frozen PVs", e);
+          }
+        },
+        FROZEN_CHECK_SECONDS,
+        FROZEN_CHECK_SECONDS,
+        TimeUnit.SECONDS);
+
+    LOGGER.log(Level.INFO, "Checking for frozen PVs every {0} s", FROZEN_CHECK_SECONDS);
+  }
+
   private void registerContextListeners(CAJContext c) throws CAException {
     c.addContextExceptionListener(
         new ContextExceptionListener() {