Timeout and override post support as property values

src/main/java/org/phoebus/channelfinder/processors/aa/AAChannelProcessor.java

@@ -7,6 +7,7 @@
 import org.phoebus.channelfinder.processors.ChannelProcessor;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.beans.factory.annotation.Autowired;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -51,6 +52,7 @@ public class AAChannelProcessor implements ChannelProcessor {
     @Value("${aa.auto_pause:}")
     private List<String> autoPauseOptions;
 
+    @Autowired
     private final ArchiverClient archiverClient = new ArchiverClient();
 
     @Override
@@ -195,24 +197,16 @@ private Map<ArchiveAction, List<ArchivePVOptions>> getArchiveActions(
         if (archivePVS.isEmpty()) {
             return result;
         }
-
-        try {
-            List<Map<String, String>> statuses = archiverClient.getStatuses(archivePVS, archiverInfo.url(), archiverInfo.version());
-            statuses
-                    .forEach(archivePVStatusJsonMap -> {
-                        String archiveStatus = archivePVStatusJsonMap.get("status");
-                        String pvName = archivePVStatusJsonMap.get("pvName");
-                        String pvStatus = archivePVS.get(pvName).getPvStatus();
-                        ArchiveAction action = pickArchiveAction(archiveStatus, pvStatus);
-                        result.get(action).add(archivePVS.get(pvName));
-                    });
-            return result;
-
-        } catch (JsonProcessingException e) {
-            // problem collecting policies from AA, so warn and return empty list
-            logger.log(Level.WARNING, () -> "Could not get AA pv Status list: " + e.getMessage());
-            return result;
-        }
+        List<Map<String, String>> statuses = archiverClient.getStatuses(archivePVS, archiverInfo.url(), archiverInfo.alias());
+        statuses
+                .forEach(archivePVStatusJsonMap -> {
+                    String archiveStatus = archivePVStatusJsonMap.get("status");
+                    String pvName = archivePVStatusJsonMap.get("pvName");
+                    String pvStatus = archivePVS.get(pvName).getPvStatus();
+                    ArchiveAction action = pickArchiveAction(archiveStatus, pvStatus);
+                    result.get(action).add(archivePVS.get(pvName));
+                });
+        return result;
     }
 
     private ArchivePVOptions createArchivePV(

src/main/java/org/phoebus/channelfinder/processors/aa/ArchiverClient.java

@@ -8,6 +8,7 @@
 import org.springframework.stereotype.Component;
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.util.UriComponentsBuilder;
+import org.springframework.beans.factory.annotation.Value;
 import reactor.core.publisher.Mono;
 
 import java.net.URI;
@@ -27,26 +28,6 @@
 public class ArchiverClient {
     private static final Logger logger = Logger.getLogger(ArchiverClient.class.getName());
     private static final int STATUS_BATCH_SIZE = 100; // Limit comes from tomcat server maxHttpHeaderSize which by default is a header of size 8k
-    private static final List<String> AA_STATUS_ENDPOINT_ONLY_SUPPORT_QUERY_VERSION = List.of("1.1.0", 
-            "Before_JDK_12_Upgrade", 
-            "v0.0.1_SNAPSHOT_03-November-2015", 
-            "v0.0.1_SNAPSHOT_09-Oct-2018", 
-            "v0.0.1_SNAPSHOT_10-June-2017", 
-            "v0.0.1_SNAPSHOT_10-Sep-2015", 
-            "v0.0.1_SNAPSHOT_12-May-2016", 
-            "v0.0.1_SNAPSHOT_12-Oct-2016", 
-            "v0.0.1_SNAPSHOT_13-Nov-2019", 
-            "v0.0.1_SNAPSHOT_14-Jun-2018", 
-            "v0.0.1_SNAPSHOT_15-Nov-2018", 
-            "v0.0.1_SNAPSHOT_20-Sept-2016", 
-            "v0.0.1_SNAPSHOT_22-June-2016", 
-            "v0.0.1_SNAPSHOT_22-June-2017", 
-            "v0.0.1_SNAPSHOT_23-Sep-2015", 
-            "v0.0.1_SNAPSHOT_26-January-2016", 
-            "v0.0.1_SNAPSHOT_27-Nov-2017", 
-            "v0.0.1_SNAPSHOT_29-July-2015", 
-            "v0.0.1_SNAPSHOT_30-March-2016", 
-            "v0.0.1_SNAPSHOT_30-September-2021");
 
     private final WebClient client = WebClient.create();
 
@@ -55,17 +36,24 @@ public class ArchiverClient {
     private static final String PV_STATUS_RESOURCE = MGMT_RESOURCE + "/getPVStatus";
     private static final String ARCHIVER_VERSIONS_RESOURCE = MGMT_RESOURCE + "/getVersions";
     private static final ObjectMapper objectMapper = new ObjectMapper();
-    private static final int TIMEOUT_SECONDS = 15;
+
+    @Value("${aa.timeout_seconds:15}")
+    private int timeoutSeconds;
+    @Value("#{${aa.query_support_override_map:{'default': false}}}")
+    private Map<String, Boolean> querySupportOverrideMap;
 
     private Stream<List<String>> partitionSet(Set<String> pvSet, int pageSize) {
         List<String> list = new ArrayList<>(pvSet);
         return IntStream.range(0, (list.size() + pageSize - 1) / pageSize)
                 .mapToObj(i -> list.subList(i * pageSize, Math.min(pageSize * (i + 1), list.size())));
     }
 
-    List<Map<String, String>> getStatuses(Map<String, ArchivePVOptions> archivePVS, String archiverURL, String archiverVersion) throws JsonProcessingException {
+    List<Map<String, String>> getStatuses(Map<String, ArchivePVOptions> archivePVS, String archiverURL, String archiverAlias) {
         Set<String> pvs = archivePVS.keySet();
-        if (AA_STATUS_ENDPOINT_ONLY_SUPPORT_QUERY_VERSION.contains(archiverVersion)) {
+        Boolean querySupportOverride = querySupportOverrideMap.getOrDefault(archiverAlias, false);
+        logger.log(Level.INFO, "Query Support Override Map: {0}", querySupportOverrideMap);
+
+        if (Boolean.TRUE.equals(querySupportOverride)) {
 
             Stream<List<String>> stream = partitionSet(pvs, STATUS_BATCH_SIZE);
 
@@ -86,7 +74,7 @@ private List<Map<String, String>> getStatusesFromPvListQuery(String archiverURL,
                 .uri(pvStatusURI)
                 .retrieve()
                 .bodyToMono(String.class)
-                .timeout(Duration.of(TIMEOUT_SECONDS, ChronoUnit.SECONDS))
+                .timeout(Duration.of(timeoutSeconds, ChronoUnit.SECONDS))
                 .onErrorResume(e -> showError(uriString, e))
                 .block();
 
@@ -96,8 +84,6 @@ private List<Map<String, String>> getStatusesFromPvListQuery(String archiverURL,
                     });
         } catch (JsonProcessingException e) {
             logger.log(Level.WARNING, "Could not parse pv status response: " + e.getMessage());
-        } catch (Exception e) {
-            logger.log(Level.WARNING, String.format("Error when trying to get status from pv list query: %s", e.getMessage()));
         }
         return List.of();
     }
@@ -110,7 +96,7 @@ private List<Map<String, String>> getStatusesFromPvListBody(String archiverURL,
                     .bodyValue(pvs)
                     .retrieve()
                     .bodyToMono(String.class)
-                    .timeout(Duration.of(TIMEOUT_SECONDS, ChronoUnit.SECONDS))
+                    .timeout(Duration.of(timeoutSeconds, ChronoUnit.SECONDS))
                     .onErrorResume(e -> showError(uriString, e))
                     .block();
 
@@ -124,8 +110,6 @@ private List<Map<String, String>> getStatusesFromPvListBody(String archiverURL,
                     });
         } catch (JsonProcessingException e) {
             logger.log(Level.WARNING, "Could not parse pv status response: " + e.getMessage());
-        } catch (Exception e) {
-            logger.log(Level.WARNING, String.format("Error when trying to get status from pv list query: %s", e.getMessage()));
         }
         return List.of();
     }
@@ -139,7 +123,7 @@ private void submitAction(String values, String endpoint, String aaURL) {
                     .bodyValue(values)
                     .retrieve()
                     .bodyToMono(String.class)
-                    .timeout(Duration.of(TIMEOUT_SECONDS, ChronoUnit.SECONDS))
+                    .timeout(Duration.of(timeoutSeconds, ChronoUnit.SECONDS))
                     .onErrorResume(e -> showError(uriString, e))
                     .block();
             logger.log(Level.FINE, () -> response);
@@ -227,7 +211,7 @@ String getVersion(String archiverURL) {
                     .uri(URI.create(uriString))
                     .retrieve()
                     .bodyToMono(String.class)
-                    .timeout(Duration.of(TIMEOUT_SECONDS, ChronoUnit.SECONDS))
+                    .timeout(Duration.of(timeoutSeconds, ChronoUnit.SECONDS))
                     .onErrorResume(e -> showError(uriString, e))
                     .block();
             Map<String, String> versionMap = objectMapper.readValue(response, Map.class);

src/main/resources/application.properties

@@ -122,6 +122,8 @@ aa.enabled=true
 aa.pva=false
 aa.archive_property_name=archive
 aa.archiver_property_name=archiver
+aa.timeout_seconds=15
+aa.query_support_override_map={'default': true}
 
 # Set the auto pause behaviour
 #

src/site/sphinx/aa_processor.rst

@@ -11,6 +11,11 @@ A list of archiver appliance URLs and aliases. ::
 
     aa.urls={'default': 'http://archiver-01.example.com:17665', 'neutron-controls': 'http://archiver-02.example.com:17665'}
 
+By default the listed archivers will get statuses by pv list body. If you are using an older archiver, this may cause data buffer limit exception: limit on max bytes to buffer.
+To set the archivers to use get statuses by pv query (legacy functionality), set the property :ref:`aa.query_support_override_map` to a map of archiver aliases to true. ::
+
+    aa.query_support_override_map={'default': true, 'neutron-controls': true}
+
 To set the choice of default archiver appliance, set the property :ref:`aa.default_alias` to the alias of the default archiver appliance. This setting can also be a comma-separated list if you want multiple default archivers.
 
 To pass the PV as "pva://PVNAME" to the archiver appliance, set the property :ref:`aa.pva` to **true**.