Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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();

Expand All @@ -55,23 +36,32 @@ 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.post_support:}")
private List<String> postSupportArchivers;

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 postSupportOverride = postSupportArchivers.contains(archiverAlias);
logger.log(Level.INFO, "Archiver Alias: {0}", archiverAlias);
logger.log(Level.INFO, "Post Support Override Archivers: {0}", postSupportArchivers);

if (Boolean.TRUE.equals(postSupportOverride)) {
logger.log(Level.INFO, "Post Support");
return getStatusesFromPvListBody(archiverURL, pvs.stream().toList());
} else {
logger.log(Level.INFO, "Query Support");
Stream<List<String>> stream = partitionSet(pvs, STATUS_BATCH_SIZE);

return stream.map(pvList -> getStatusesFromPvListQuery(archiverURL, pvList)).flatMap(List::stream).toList();
} else {
return getStatusesFromPvListBody(archiverURL, pvs.stream().toList());
}
}

Expand All @@ -86,7 +76,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();

Expand All @@ -110,7 +100,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();

Expand All @@ -125,7 +115,7 @@ 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()));
logger.log(Level.WARNING, String.format("Error when trying to get status from pv list body: %s", e.getMessage()));
}
return List.of();
}
Expand All @@ -139,7 +129,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);
Expand Down Expand Up @@ -227,7 +217,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);
Expand Down
4 changes: 4 additions & 0 deletions src/main/resources/application.properties
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,10 @@ aa.enabled=true
aa.pva=false
aa.archive_property_name=archive
aa.archiver_property_name=archiver
aa.timeout_seconds=15

# Comma-separated list of archivers to use post support
aa.post_support=

# Set the auto pause behaviour
#
Expand Down
5 changes: 5 additions & 0 deletions src/site/sphinx/aa_processor.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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 query.
To set the archivers to use get statuses by pv list body, set the property :ref:`aa.post_support` to a comma separated list of archiver aliases. ::

aa.post_support=default, neutron-controls

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**.
Expand Down
Loading
Loading