Skip to content

Commit bf01d66

Browse files
jacomagoclaude
jacomago
and
claude
committed
feat(aa): cache AA policies with scheduled background refresh
Policies are fetched once at startup (@PostConstruct) and refreshed on a configurable fixed delay (aa.policy_refresh_interval_seconds, default 1 h) rather than on every process() call. process() now reads a volatile snapshot, so policy fetches can no longer block or fail mid-run. processorInfo() exposes LastPolicyRefresh and per-archiver policy counts for observability. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent 1c8276c commit bf01d66

13 files changed

Lines changed: 264 additions & 34 deletions

‎src/main/java/org/phoebus/channelfinder/configuration/AAChannelProcessor.java‎

Lines changed: 99 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,16 @@
11
package org.phoebus.channelfinder.configuration;
22

3+
import jakarta.annotation.PostConstruct;
4+
import java.time.Instant;
35
import java.util.ArrayList;
46
import java.util.Arrays;
7+
import java.util.Collections;
58
import java.util.EnumMap;
69
import java.util.HashMap;
710
import java.util.List;
811
import java.util.Map;
912
import java.util.Optional;
13+
import java.util.concurrent.TimeUnit;
1014
import java.util.logging.Level;
1115
import java.util.logging.Logger;
1216
import java.util.stream.Collectors;
@@ -20,10 +24,12 @@
2024
import org.phoebus.channelfinder.service.model.archiver.aa.ArchiveAction;
2125
import org.phoebus.channelfinder.service.model.archiver.aa.ArchivePVOptions;
2226
import org.phoebus.channelfinder.service.model.archiver.aa.ArchiverInfo;
27+
import org.phoebus.channelfinder.service.model.archiver.aa.ArchiverPolicies;
2328
import org.springframework.beans.factory.annotation.Autowired;
2429
import org.springframework.beans.factory.annotation.Value;
2530
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
2631
import org.springframework.context.annotation.Configuration;
32+
import org.springframework.scheduling.annotation.Scheduled;
2733
import tools.jackson.core.JacksonException;
2834

2935
/**
@@ -70,6 +76,12 @@ public class AAChannelProcessor implements ChannelProcessor {
7076

7177
@Autowired private ArchiverService archiverService;
7278

79+
@Value("${aa.policy_refresh_interval_seconds:3600}")
80+
private long policyRefreshIntervalSeconds;
81+
82+
private volatile Map<String, ArchiverPolicies> cachedPolicies = Map.of();
83+
private volatile Instant lastPolicyRefresh;
84+
7385
@Override
7486
public boolean enabled() {
7587
return aaEnabled;
@@ -80,8 +92,24 @@ public void setEnabled(boolean enabled) {
8092
this.aaEnabled = enabled;
8193
}
8294

95+
@PostConstruct
96+
public void initPolicyCache() {
97+
refreshPolicies();
98+
}
99+
100+
@Scheduled(
101+
fixedDelayString = "${aa.policy_refresh_interval_seconds:3600}",
102+
timeUnit = TimeUnit.SECONDS)
103+
public void scheduledPolicyRefresh() {
104+
refreshPolicies();
105+
}
106+
83107
@Override
84108
public ChannelProcessorInfo processorInfo() {
109+
String policyCounts =
110+
cachedPolicies.entrySet().stream()
111+
.map(e -> e.getKey() + "=" + e.getValue())
112+
.collect(Collectors.joining(", "));
85113
return new ChannelProcessorInfo(
86114
"AAChannelProcessor",
87115
aaEnabled,
@@ -93,7 +121,15 @@ public ChannelProcessorInfo processorInfo() {
93121
"Archivers",
94122
aaURLs.keySet().toString(),
95123
"AutoPauseOn",
96-
autoPauseOptions.toString()));
124+
autoPauseOptions.toString(),
125+
"PostSupportArchivers",
126+
postSupportArchivers.toString(),
127+
"PolicyRefreshIntervalSeconds",
128+
String.valueOf(policyRefreshIntervalSeconds),
129+
"LastPolicyRefresh",
130+
lastPolicyRefresh == null ? "never" : lastPolicyRefresh.toString(),
131+
"CachedPoliciesPerArchiver",
132+
policyCounts.isEmpty() ? "none" : policyCounts));
97133
}
98134

99135
/**
@@ -112,7 +148,7 @@ public long process(List<Channel> channels) throws JacksonException {
112148
return 0;
113149
}
114150

115-
Map<String, ArchiverInfo> archiversInfo = getArchiversInfo(aaURLs);
151+
Map<String, ArchiverInfo> archiversInfo = getArchiversInfoFromCache();
116152
if (archiversInfo.isEmpty()) {
117153
logger.log(
118154
Level.WARNING,
@@ -271,10 +307,6 @@ private ArchiveAction pickArchiveAction(String archiveStatus, String pvStatus) {
271307

272308
private Optional<Map<ArchiveAction, List<ArchivePVOptions>>> getArchiveActions(
273309
Map<String, ArchivePVOptions> archivePVS, ArchiverInfo archiverInfo) {
274-
if (archiverInfo == null) {
275-
return Optional.of(Map.of());
276-
}
277-
278310
logger.log(
279311
Level.FINE,
280312
() ->
@@ -358,14 +390,71 @@ private ArchivePVOptions createArchivePV(
358390
return newArchiverPV;
359391
}
360392

361-
private Map<String, ArchiverInfo> getArchiversInfo(Map<String, String> aaURLs) {
393+
private Map<String, ArchiverInfo> getArchiversInfoFromCache() {
394+
Map<String, ArchiverPolicies> snapshot = cachedPolicies;
362395
return aaURLs.entrySet().stream()
363-
.filter(aa -> !StringUtils.isEmpty(aa.getValue()))
396+
.filter(e -> !StringUtils.isEmpty(e.getValue()))
397+
.filter(
398+
e -> {
399+
if (!snapshot.containsKey(e.getKey())) {
400+
logger.log(
401+
Level.WARNING,
402+
() ->
403+
String.format(
404+
"Archiver '%s' has no cached policies (unreachable at last refresh); skipping.",
405+
e.getKey()));
406+
return false;
407+
}
408+
return true;
409+
})
364410
.collect(
365411
Collectors.toMap(
366412
Map.Entry::getKey,
367-
aa ->
413+
e ->
368414
new ArchiverInfo(
369-
aa.getKey(), aa.getValue(), archiverService.getAAPolicies(aa.getValue()))));
415+
e.getKey(), e.getValue(), snapshot.get(e.getKey()).policies())));
416+
}
417+
418+
private void refreshPolicies() {
419+
if (aaURLs.isEmpty()) {
420+
logger.log(Level.FINE, "No archivers configured; skipping policy cache refresh.");
421+
return;
422+
}
423+
Map<String, ArchiverPolicies> current = cachedPolicies;
424+
Map<String, ArchiverPolicies> updated = new HashMap<>(current);
425+
List<String> changed = new ArrayList<>();
426+
427+
for (Map.Entry<String, String> entry : aaURLs.entrySet()) {
428+
if (StringUtils.isEmpty(entry.getValue())) continue;
429+
String alias = entry.getKey();
430+
try {
431+
List<String> fresh = archiverService.getAAPolicies(entry.getValue());
432+
ArchiverPolicies existing = current.get(alias);
433+
if (existing == null || !existing.policies().equals(fresh)) {
434+
updated.put(alias, new ArchiverPolicies(fresh));
435+
changed.add(alias);
436+
}
437+
} catch (ArchiverServiceException ex) {
438+
logger.log(
439+
Level.WARNING,
440+
() ->
441+
String.format(
442+
"Policy fetch failed for archiver '%s'; retaining cached policies: %s",
443+
alias, ex.getMessage()));
444+
}
445+
}
446+
447+
lastPolicyRefresh = Instant.now();
448+
if (!changed.isEmpty()) {
449+
cachedPolicies = Collections.unmodifiableMap(updated);
450+
logger.log(
451+
Level.INFO,
452+
"AA policy cache updated: {0}",
453+
cachedPolicies.entrySet().stream()
454+
.map(e -> e.getKey() + "=" + e.getValue())
455+
.collect(Collectors.joining(", ")));
456+
} else {
457+
logger.log(Level.FINE, "AA policy cache unchanged after refresh.");
458+
}
370459
}
371460
}

‎src/main/java/org/phoebus/channelfinder/service/external/ArchiverService.java‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -237,15 +237,14 @@ public List<String> getAAPolicies(String aaURL) {
237237
.retrieve()
238238
.body(new ParameterizedTypeReference<>() {});
239239
if (policyMap == null) {
240-
return List.of();
240+
throw new ArchiverServiceException("Null response from " + uriString);
241241
}
242242
return new ArrayList<>(policyMap.keySet());
243+
} catch (ArchiverServiceException e) {
244+
throw e;
243245
} catch (Exception e) {
244-
// problem collecting policies from AA, so warn and return empty list
245-
logger.log(
246-
Level.WARNING,
247-
() -> "Could not get AA policies list from " + aaURL + ": " + e.getMessage());
248-
return List.of();
246+
throw new ArchiverServiceException(
247+
"Could not get AA policies from " + aaURL + ": " + e.getMessage(), e);
249248
}
250249
}
251250
}

‎src/main/java/org/phoebus/channelfinder/service/model/archiver/aa/ArchiverPolicies.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
package org.phoebus.channelfinder.service.model.archiver.aa;
2+
3+
import java.util.List;
4+
5+
public record ArchiverPolicies(List<String> policies) {
6+
@Override
7+
public String toString() {
8+
return policies.size() + " policies";
9+
}
10+
}

‎src/main/resources/application.properties‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,8 @@ aa.pva=false
143143
aa.archive_property_name=archive
144144
aa.archiver_property_name=archiver
145145
aa.timeout_seconds=15
146+
# Interval in seconds between automatic AA policy cache refreshes (default: 1 hour)
147+
aa.policy_refresh_interval_seconds=3600
146148

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

‎src/test/java/org/phoebus/channelfinder/processors/aa/AAChannelProcessorIT.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import java.util.Map;
1515
import java.util.stream.Stream;
1616
import org.jetbrains.annotations.NotNull;
17+
import org.junit.jupiter.api.BeforeEach;
1718
import org.junit.jupiter.api.Test;
1819
import org.junit.jupiter.api.extension.ExtendWith;
1920
import org.junit.jupiter.params.ParameterizedTest;
@@ -47,6 +48,12 @@ class AAChannelProcessorIT {
4748
@MockitoBean ArchiverService archiverService;
4849
@Autowired AAChannelProcessor aaChannelProcessor;
4950

51+
@BeforeEach
52+
void primeCache() {
53+
when(archiverService.getAAPolicies(anyString())).thenReturn(List.of("policy"));
54+
aaChannelProcessor.scheduledPolicyRefresh();
55+
}
56+
5057
@NotNull
5158
private static Stream<Arguments> processSource() {
5259
return Stream.of(
@@ -103,9 +110,6 @@ public static void paramableAAChannelProcessorTest(
103110
String archiveStatus,
104111
String archiverEndpoint)
105112
throws JacksonException {
106-
// Mock getAAPolicies
107-
when(archiverService.getAAPolicies(anyString())).thenReturn(List.of("policy"));
108-
109113
if (!archiveStatus.isEmpty()) {
110114
// Mock getStatuses
111115
List<Map<String, String>> archivePVStatuses =
@@ -126,8 +130,6 @@ public static void paramableAAChannelProcessorTest(
126130
assertEquals(count, archiverEndpoint.isEmpty() ? 0 : channels.size());
127131

128132
// Verifications
129-
verify(archiverService).getAAPolicies(anyString());
130-
131133
if (!archiveStatus.isEmpty()) {
132134
verify(archiverService).getStatusesViaGet(anyString(), anyList());
133135
}

‎src/test/java/org/phoebus/channelfinder/processors/aa/AAChannelProcessorMultiArchiverIT.java‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@
33
import static org.mockito.ArgumentMatchers.anyList;
44
import static org.mockito.ArgumentMatchers.anyMap;
55
import static org.mockito.ArgumentMatchers.anyString;
6-
import static org.mockito.Mockito.times;
76
import static org.mockito.Mockito.verify;
87
import static org.mockito.Mockito.when;
98
import static org.phoebus.channelfinder.processors.aa.AAChannelProcessorIT.activeProperty;
@@ -15,6 +14,7 @@
1514
import java.util.stream.Collectors;
1615
import java.util.stream.IntStream;
1716
import java.util.stream.Stream;
17+
import org.junit.jupiter.api.BeforeEach;
1818
import org.junit.jupiter.params.ParameterizedTest;
1919
import org.junit.jupiter.params.provider.Arguments;
2020
import org.junit.jupiter.params.provider.MethodSource;
@@ -39,6 +39,12 @@ class AAChannelProcessorMultiArchiverIT {
3939
@Autowired AAChannelProcessor aaChannelProcessor;
4040
@MockitoBean ArchiverService archiverService;
4141

42+
@BeforeEach
43+
void primeCache() {
44+
when(archiverService.getAAPolicies(anyString())).thenReturn(List.of("policy"));
45+
aaChannelProcessor.scheduledPolicyRefresh();
46+
}
47+
4248
static Stream<Arguments> provideArguments() {
4349
List<Channel> channels =
4450
List.of(
@@ -99,8 +105,6 @@ void testProcessMultiArchivers(
99105
Map<String, String> namesToStatuses,
100106
Map<ArchiveAction, List<String>> actionsToNames)
101107
throws JacksonException {
102-
when(archiverService.getAAPolicies(anyString())).thenReturn(List.of("policy"));
103-
104108
// Request to archiver status
105109
List<Map<String, String>> archivePVStatuses =
106110
namesToStatuses.entrySet().stream()
@@ -116,9 +120,7 @@ void testProcessMultiArchivers(
116120

117121
aaChannelProcessor.process(channels);
118122

119-
// Verifications
120-
verify(archiverService, times(2)).getAAPolicies(anyString());
121-
123+
// Verifications: query archiver uses GET, post archiver uses POST (aa.post_support=post)
122124
if (!namesToStatuses.isEmpty()) {
123125
verify(archiverService).getStatusesViaGet(anyString(), anyList());
124126
verify(archiverService).getStatusesViaPost(anyString(), anyList());

‎src/test/java/org/phoebus/channelfinder/processors/aa/AAChannelProcessorMultiIT.java‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import java.util.stream.Collectors;
1717
import java.util.stream.IntStream;
1818
import java.util.stream.Stream;
19+
import org.junit.jupiter.api.BeforeEach;
1920
import org.junit.jupiter.params.ParameterizedTest;
2021
import org.junit.jupiter.params.provider.Arguments;
2122
import org.junit.jupiter.params.provider.MethodSource;
@@ -42,6 +43,12 @@ class AAChannelProcessorMultiIT {
4243
@Autowired AAChannelProcessor aaChannelProcessor;
4344
@MockitoBean ArchiverService archiverService;
4445

46+
@BeforeEach
47+
void primeCache() {
48+
when(archiverService.getAAPolicies(anyString())).thenReturn(List.of("policy"));
49+
aaChannelProcessor.scheduledPolicyRefresh();
50+
}
51+
4552
static Stream<Arguments> provideArguments() {
4653
List<Channel> channels =
4754
List.of(
@@ -104,9 +111,6 @@ void testProcessMulti(
104111
int expectedProcessedChannels)
105112
throws JacksonException {
106113

107-
// Mock getAAPolicies
108-
when(archiverService.getAAPolicies(anyString())).thenReturn(List.of("policy"));
109-
110114
// Mock getStatuses
111115
List<Map<String, String>> archivePVStatuses =
112116
namesToStatuses.entrySet().stream()
@@ -121,7 +125,6 @@ void testProcessMulti(
121125
long count = aaChannelProcessor.process(channels);
122126
assertEquals(expectedProcessedChannels, count);
123127

124-
verify(archiverService).getAAPolicies(anyString());
125128
verify(archiverService).getStatusesViaGet(anyString(), anyList());
126129

127130
ArgumentCaptor<Map<ArchiveAction, List<ArchivePVOptions>>> captor =

‎src/test/java/org/phoebus/channelfinder/processors/aa/AAChannelProcessorNoDefaultIT.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,14 @@
11
package org.phoebus.channelfinder.processors.aa;
22

3+
import static org.mockito.ArgumentMatchers.anyString;
4+
import static org.mockito.Mockito.when;
35
import static org.phoebus.channelfinder.processors.aa.AAChannelProcessorIT.activeProperty;
46
import static org.phoebus.channelfinder.processors.aa.AAChannelProcessorIT.archiveProperty;
57
import static org.phoebus.channelfinder.processors.aa.AAChannelProcessorIT.paramableAAChannelProcessorTest;
68

79
import java.util.List;
810
import java.util.stream.Stream;
11+
import org.junit.jupiter.api.BeforeEach;
912
import org.junit.jupiter.params.ParameterizedTest;
1013
import org.junit.jupiter.params.provider.Arguments;
1114
import org.junit.jupiter.params.provider.MethodSource;
@@ -30,6 +33,12 @@ class AAChannelProcessorNoDefaultIT {
3033

3134
@MockitoBean ArchiverService archiverService;
3235

36+
@BeforeEach
37+
void primeCache() {
38+
when(archiverService.getAAPolicies(anyString())).thenReturn(List.of("policy"));
39+
aaChannelProcessor.scheduledPolicyRefresh();
40+
}
41+
3342
private static Stream<Arguments> processNoPauseSource() {
3443

3544
return Stream.of(

0 commit comments

Comments
 (0)