Skip to content

Commit

Permalink
fix deleteRecording mutation bug
Browse files Browse the repository at this point in the history
  • Loading branch information
@aali309
aali309 committed May 4, 2024
1 parent e97de33 commit e9b4de4
Show file tree
Hide file tree
Showing 2 changed files with 168 additions and 229 deletions.
236 changes: 133 additions & 103 deletions src/main/java/io/cryostat/graphql/ActiveRecordings.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import java.util.Objects;
import java.util.Optional;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import org.openjdk.jmc.common.unit.QuantityConversionException;

Expand All @@ -45,6 +46,7 @@
import io.smallrye.common.annotation.Blocking;
import io.smallrye.graphql.api.Nullable;
import io.smallrye.graphql.execution.ExecutionException;
import io.smallrye.mutiny.Uni;
import jakarta.inject.Inject;
import jakarta.transaction.Transactional;
import jdk.jfr.RecordingState;
Expand All @@ -59,33 +61,32 @@
@GraphQLApi
public class ActiveRecordings {

@Inject RecordingHelper recordingHelper;
@Inject Logger logger;
@Inject
RecordingHelper recordingHelper;
@Inject
Logger logger;

@ConfigProperty(name = ConfigProperties.CONNECTIONS_FAILED_TIMEOUT)
Duration timeout;

@Blocking
@Transactional
@Mutation
@Description(
"Start a new Flight Recording on all Targets under the subtrees of the discovery nodes"
+ " matching the given filter")
@Description("Start a new Flight Recording on all Targets under the subtrees of the discovery nodes"
+ " matching the given filter")
public List<ActiveRecording> createRecording(
@NonNull DiscoveryNodeFilter nodes, @NonNull RecordingSettings recording) {
return DiscoveryNode.<DiscoveryNode>listAll().stream()
.filter(nodes)
.flatMap(
node ->
RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
node -> RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
.map(
target -> {
var template =
recordingHelper.getPreferredTemplate(
target,
recording.template,
TemplateType.valueOf(recording.templateType));
var template = recordingHelper.getPreferredTemplate(
target,
recording.template,
TemplateType.valueOf(recording.templateType));
try {
return recordingHelper
.startRecording(
Expand All @@ -110,21 +111,18 @@ public List<ActiveRecording> createRecording(
@Blocking
@Transactional
@Mutation
@Description(
"Archive an existing Flight Recording matching the given filter, on all Targets under"
+ " the subtrees of the discovery nodes matching the given filter")
@Description("Archive an existing Flight Recording matching the given filter, on all Targets under"
+ " the subtrees of the discovery nodes matching the given filter")
public List<ArchivedRecording> archiveRecording(
@NonNull DiscoveryNodeFilter nodes, @Nullable ActiveRecordingsFilter recordings) {
return DiscoveryNode.<DiscoveryNode>listAll().stream()
.filter(nodes)
.flatMap(
node ->
RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
node -> RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
.flatMap(
t ->
recordingHelper.listActiveRecordings(t).stream()
.filter(r -> recordings == null || recordings.test(r)))
t -> recordingHelper.listActiveRecordings(t).stream()
.filter(r -> recordings == null || recordings.test(r)))
.map(
recording -> {
try {
Expand All @@ -139,21 +137,18 @@ public List<ArchivedRecording> archiveRecording(
@Blocking
@Transactional
@Mutation
@Description(
"Stop an existing Flight Recording matching the given filter, on all Targets under"
+ " the subtrees of the discovery nodes matching the given filter")
@Description("Stop an existing Flight Recording matching the given filter, on all Targets under"
+ " the subtrees of the discovery nodes matching the given filter")
public List<ActiveRecording> stopRecording(
@NonNull DiscoveryNodeFilter nodes, @Nullable ActiveRecordingsFilter recordings) {
return DiscoveryNode.<DiscoveryNode>listAll().stream()
.filter(nodes)
.flatMap(
node ->
RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
node -> RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
.flatMap(
t ->
recordingHelper.listActiveRecordings(t).stream()
.filter(r -> recordings == null || recordings.test(r)))
t -> recordingHelper.listActiveRecordings(t).stream()
.filter(r -> recordings == null || recordings.test(r)))
.map(
recording -> {
try {
Expand All @@ -171,48 +166,92 @@ public List<ActiveRecording> stopRecording(
@Blocking
@Transactional
@Mutation
@Description(
"Delete an existing Flight Recording matching the given filter, on all Targets under"
+ " the subtrees of the discovery nodes matching the given filter")
public List<ActiveRecording> deleteRecording(
@Description("Delete an existing Flight Recording matching the given filter, on all Targets under"
+ " the subtrees of the discovery nodes matching the given filter")
public Uni<List<ActiveRecording>> deleteRecording(
@NonNull DiscoveryNodeFilter nodes, @Nullable ActiveRecordingsFilter recordings) {
return DiscoveryNode.<DiscoveryNode>listAll().stream()
.filter(nodes)
.flatMap(
node ->
RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
.flatMap(
t ->
recordingHelper.listActiveRecordings(t).stream()
.filter(r -> recordings == null || recordings.test(r)))
.map(
recording -> {
try {
return recordingHelper
.deleteRecording(recording)
.await()
.atMost(Duration.ofSeconds(10));
} catch (Exception e) {
throw new ExecutionException(e);
}
})
.toList();

Logger logger = Logger.getLogger(ActiveRecordings.class);

logger.info("Starting the deletion of recordings");

// Fetch all recordings to delete based on filters
return Uni.createFrom()
.item(
() -> DiscoveryNode.<DiscoveryNode>listAll().stream()
.filter(nodes)
.peek(
node -> logger.debug(
"++Filtering node: " + node.name))
.flatMap(
node -> RootNode.recurseChildren(
node, n -> n.target != null)
.stream())
.peek(
n -> logger.debug(
"++Processing child node with"
+ " target: "
+ n.target.connectUrl))
.map(n -> n.target)
.peek(
target -> logger.debug(
"++Target to delete recordings"
+ " from: "
+ target.connectUrl))
.flatMap(
target -> recordingHelper
.listActiveRecordings(target)
.stream())
.filter(r -> recordings == null || recordings.test(r))
.collect(Collectors.toList()))
.onItem()
.transformToUni(
recordingsToDelete -> {
logger.info(
"++Recordings collected for deletion: "
+ recordingsToDelete.size());

List<Uni<ActiveRecording>> deletionUnis = recordingsToDelete.stream()
.map(
recording -> recordingHelper
.deleteRecording(recording)
.onItem()
.invoke(
deleted -> logger.info(
"++Deleted"
+ " recording:"
+ " "
+ deleted.name))
.onFailure()
.recoverWithItem(
(ActiveRecording) null))
.collect(Collectors.toList());

return Uni.combine()
.all()
.unis(deletionUnis)
.combinedWith(
results -> {
// Filter nulls to avoid failed deletions
return results.stream()
.filter(Objects::nonNull)
.map(ActiveRecording.class::cast)
.collect(Collectors.toList());
});
});
}

@Blocking
@Transactional
@Mutation
@Description(
"Create a Flight Recorder Snapshot on all Targets under"
+ " the subtrees of the discovery nodes matching the given filter")
@Description("Create a Flight Recorder Snapshot on all Targets under"
+ " the subtrees of the discovery nodes matching the given filter")
public List<ActiveRecording> createSnapshot(@NonNull DiscoveryNodeFilter nodes) {
return DiscoveryNode.<DiscoveryNode>listAll().stream()
.filter(nodes)
.flatMap(
node ->
RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
node -> RootNode.recurseChildren(node, n -> n.target != null).stream()
.map(n -> n.target))
.map(
target -> {
try {
Expand All @@ -234,9 +273,8 @@ public ActiveRecording doStartRecording(
@Source Target target, @NonNull RecordingSettings recording)
throws QuantityConversionException {
var fTarget = Target.getTargetById(target.id);
Template template =
recordingHelper.getPreferredTemplate(
fTarget, recording.template, TemplateType.valueOf(recording.templateType));
Template template = recordingHelper.getPreferredTemplate(
fTarget, recording.template, TemplateType.valueOf(recording.templateType));
return recordingHelper
.startRecording(
fTarget,
Expand Down Expand Up @@ -289,8 +327,7 @@ public TargetNodes.ActiveRecordings active(

var in = recordings.active;
if (in != null && in.data != null) {
out.data =
in.data.stream().filter(r -> filter == null ? true : filter.test(r)).toList();
out.data = in.data.stream().filter(r -> filter == null ? true : filter.test(r)).toList();
out.aggregate = AggregateInfo.fromActive(out.data);
}

Expand Down Expand Up @@ -335,7 +372,8 @@ public static class MetadataLabels {

private Map<String, String> labels;

public MetadataLabels() {}
public MetadataLabels() {
}

public MetadataLabels(Map<String, String> labels) {
this.labels = new HashMap<>(labels);
Expand Down Expand Up @@ -370,44 +408,36 @@ public static class ActiveRecordingsFilter implements Predicate<ActiveRecording>

@Override
public boolean test(ActiveRecording r) {
Predicate<ActiveRecording> matchesName =
n -> name == null || Objects.equals(name, n.name);
Predicate<ActiveRecording> matchesName = n -> name == null || Objects.equals(name, n.name);
Predicate<ActiveRecording> matchesNames = n -> names == null || names.contains(n.name);
Predicate<ActiveRecording> matchesLabels =
n ->
labels == null
|| labels.stream()
.allMatch(
label ->
LabelSelectorMatcher.parse(label)
.test(n.metadata.labels()));
Predicate<ActiveRecording> matchesLabels = n -> labels == null
|| labels.stream()
.allMatch(
label -> LabelSelectorMatcher.parse(label)
.test(n.metadata.labels()));
Predicate<ActiveRecording> matchesState = n -> state == null || n.state.equals(state);
Predicate<ActiveRecording> matchesContinuous =
n -> continuous == null || continuous.equals(n.continuous);
Predicate<ActiveRecording> matchesToDisk =
n -> toDisk == null || toDisk.equals(n.toDisk);
Predicate<ActiveRecording> matchesDurationGte =
n ->
durationMsGreaterThanEqual == null
|| durationMsGreaterThanEqual >= n.duration;
Predicate<ActiveRecording> matchesDurationLte =
n -> durationMsLessThanEqual == null || durationMsLessThanEqual <= n.duration;
Predicate<ActiveRecording> matchesStartTimeAfter =
n -> startTimeMsAfterEqual == null || startTimeMsAfterEqual >= n.startTime;
Predicate<ActiveRecording> matchesStartTimeBefore =
n -> startTimeMsBeforeEqual == null || startTimeMsBeforeEqual <= n.startTime;
Predicate<ActiveRecording> matchesContinuous = n -> continuous == null || continuous.equals(n.continuous);
Predicate<ActiveRecording> matchesToDisk = n -> toDisk == null || toDisk.equals(n.toDisk);
Predicate<ActiveRecording> matchesDurationGte = n -> durationMsGreaterThanEqual == null
|| durationMsGreaterThanEqual >= n.duration;
Predicate<ActiveRecording> matchesDurationLte = n -> durationMsLessThanEqual == null
|| durationMsLessThanEqual <= n.duration;
Predicate<ActiveRecording> matchesStartTimeAfter = n -> startTimeMsAfterEqual == null
|| startTimeMsAfterEqual >= n.startTime;
Predicate<ActiveRecording> matchesStartTimeBefore = n -> startTimeMsBeforeEqual == null
|| startTimeMsBeforeEqual <= n.startTime;

return List.of(
matchesName,
matchesNames,
matchesLabels,
matchesState,
matchesContinuous,
matchesToDisk,
matchesDurationGte,
matchesDurationLte,
matchesStartTimeBefore,
matchesStartTimeAfter)
matchesName,
matchesNames,
matchesLabels,
matchesState,
matchesContinuous,
matchesToDisk,
matchesDurationGte,
matchesDurationLte,
matchesStartTimeBefore,
matchesStartTimeAfter)
.stream()
.reduce(x -> true, Predicate::and)
.test(r);
Expand Down
Loading

0 comments on commit e9b4de4

Please sign in to comment.