fix(discovery): harden Discovery Plugin registration against failed plugins, retry floods

src/main/java/io/cryostat/ConfigProperties.java

@@ -50,6 +50,12 @@ public class ConfigProperties {
     public static final String CONTAINERS_POLL_PERIOD = "cryostat.discovery.containers.poll-period";
     public static final String CONTAINERS_REQUEST_TIMEOUT =
             "cryostat.discovery.containers.request-timeout";
+    public static final String DISCOVERY_PLUGINS_MAX_FAILURES =
+            "cryostat.discovery.plugins.max-failures";
+    public static final String DISCOVERY_PLUGINS_PING_PERIOD =
+            "cryostat.discovery.plugins.ping-period";
+    public static final String DISCOVERY_PLUGINS_MAX_BACKOFF_MULTIPLIER =
+            "cryostat.discovery.plugins.max-backoff-multiplier";
 
     public static final String CONNECTIONS_TTL = "cryostat.connections.ttl";
     public static final String CONNECTIONS_FAILED_BACKOFF = "cryostat.connections.failed-backoff";

src/main/java/io/cryostat/discovery/CustomDiscovery.java

@@ -39,6 +39,8 @@
 import jakarta.annotation.security.RolesAllowed;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
+import jakarta.persistence.EntityManager;
+import jakarta.persistence.LockModeType;
 import jakarta.transaction.Transactional;
 import jakarta.ws.rs.BadRequestException;
 import jakarta.ws.rs.Consumes;
@@ -78,6 +80,7 @@ public class CustomDiscovery {
 
     @Inject Logger logger;
     @Inject EventBus bus;
+    @Inject EntityManager entityManager;
     @Inject TargetConnectionManager connectionManager;
     @Inject URIUtil uriUtil;
 
@@ -200,11 +203,16 @@ RestResponse<Target> doCreate(
                                         DiscoveryNode.target(target, NodeType.BaseNodeType.JVM);
                                 target.discoveryNode = node;
                                 DiscoveryNode realm = DiscoveryNode.getRealm(REALM).orElseThrow();
+                                realm =
+                                        entityManager.find(
+                                                DiscoveryNode.class,
+                                                realm.id,
+                                                LockModeType.PESSIMISTIC_WRITE);
 
-                                realm.children.add(node);
                                 node.parent = realm;
                                 target.persist();
                                 node.persist();
+                                realm.children.add(node);
                                 realm.persist();
 
                                 return ResponseBuilder.<Target>created(
@@ -239,6 +247,7 @@ RestResponse<Target> doCreate(
     public void delete(@RestPath long id) throws URISyntaxException {
         Target target = Target.find("id", id).singleResult();
         DiscoveryNode realm = DiscoveryNode.getRealm(REALM).orElseThrow();
+        realm = entityManager.find(DiscoveryNode.class, realm.id, LockModeType.PESSIMISTIC_WRITE);
         boolean withinRealm = realm.children.remove(target.discoveryNode);
         if (!withinRealm) {
             throw new BadRequestException();

src/main/java/io/cryostat/discovery/DiscoveryJwtFactory.java

@@ -35,6 +35,8 @@
 import javax.crypto.KeyGenerator;
 import javax.crypto.SecretKey;
 
+import io.cryostat.ConfigProperties;
+
 import com.nimbusds.jose.EncryptionMethod;
 import com.nimbusds.jose.JOSEException;
 import com.nimbusds.jose.JWEAlgorithm;
@@ -71,7 +73,7 @@ public class DiscoveryJwtFactory {
     static final String DISCOVERY_V4_API_PATH = "/api/v4/discovery/";
     static final String DISCOVERY_V4_2_API_PATH = "/api/v4.2/discovery/";
 
-    @ConfigProperty(name = "cryostat.discovery.plugins.ping-period")
+    @ConfigProperty(name = ConfigProperties.DISCOVERY_PLUGINS_PING_PERIOD)
     Duration discoveryPingPeriod;
 
     @ConfigProperty(name = "cryostat.discovery.plugins.jwt.signature.algorithm")

src/main/java/io/cryostat/discovery/DiscoveryPlugin.java

@@ -17,6 +17,8 @@
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.time.Instant;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.function.Supplier;
 
@@ -74,7 +76,12 @@
 @NamedQueries({
     @NamedQuery(
             name = "DiscoveryPlugin.getBuiltinRealmIds",
-            query = "SELECT p.realm.id FROM DiscoveryPlugin p WHERE p.builtin = true")
+            query = "SELECT p.realm.id FROM DiscoveryPlugin p WHERE p.builtin = true"),
+    @NamedQuery(
+            name = "DiscoveryPlugin.findByCallbackAndRealmName",
+            query =
+                    "SELECT p FROM DiscoveryPlugin p JOIN FETCH p.realm r WHERE p.callback = ?1"
+                            + " AND r.name = ?2")
 })
 public class DiscoveryPlugin extends PanacheEntityBase {
 
@@ -109,10 +116,41 @@ public class DiscoveryPlugin extends PanacheEntityBase {
     @JsonProperty(access = JsonProperty.Access.READ_ONLY)
     public boolean builtin;
 
+    @Column(nullable = false)
+    @JsonIgnore
+    public int consecutiveFailures = 0;
+
+    @Column(nullable = true)
+    @Convert(converter = InstantConverter.class)
+    @JsonIgnore
+    public Instant lastSuccessfulPing;
+
+    @Column(nullable = true)
+    @Convert(converter = InstantConverter.class)
+    @JsonIgnore
+    public Instant lastFailedPing;
+
+    @Column(nullable = false)
+    @JsonIgnore
+    public int backoffMultiplier = 1;
+
+    @Column(nullable = true)
+    @Convert(converter = InstantConverter.class)
+    @JsonIgnore
+    public Instant nextPingAt;
+
+    public static Optional<DiscoveryPlugin> findByCallbackAndRealmName(
+            URI callback, String realmName) {
+        return DiscoveryPlugin.<DiscoveryPlugin>find(
+                        "#DiscoveryPlugin.findByCallbackAndRealmName", callback, realmName)
+                .singleResultOptional();
+    }
+
     @ApplicationScoped
     static class Listener {
 
         @Inject Logger logger;
+        @Inject PluginCallbackFactory callbackFactory;
 
         @PrePersist
         @Transactional
@@ -126,13 +164,22 @@ public void prePersist(DiscoveryPlugin plugin) {
             if (plugin.credential == null) {
                 var credential = getCredential(plugin);
                 plugin.credential = credential;
+                credential.discoveryPlugin = plugin;
                 plugin.callback = UriBuilder.fromUri(plugin.callback).userInfo(null).build();
             }
+            if (plugin.nextPingAt != null
+                    || plugin.lastFailedPing != null
+                    || plugin.lastSuccessfulPing != null) {
+                logger.debugv(
+                        "Skipping prePersist ping for plugin with existing state: {0} @ {1}",
+                        plugin.realm.name, plugin.callback);
+                return;
+            }
             try {
                 logger.debugv(
                         "Testing discovery plugin callback: {0} @ {1}",
                         plugin.realm.name, plugin.callback);
-                PluginCallback.create(plugin).ping();
+                callbackFactory.create(plugin).ping();
                 logger.debugv(
                         "Registered discovery plugin: {0} @ {1}",
                         plugin.realm.name, plugin.callback);

src/main/java/io/cryostat/discovery/InstantConverter.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright The Cryostat Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.cryostat.discovery;
+
+import java.time.Instant;
+
+import jakarta.persistence.AttributeConverter;
+
+public class InstantConverter implements AttributeConverter<Instant, Long> {
+
+    @Override
+    public Long convertToDatabaseColumn(Instant attribute) {
+        if (attribute == null) {
+            return null;
+        }
+        return attribute.toEpochMilli();
+    }
+
+    @Override
+    public Instant convertToEntityAttribute(Long dbData) {
+        if (dbData == null) {
+            return null;
+        }
+        return Instant.ofEpochMilli(dbData);
+    }
+}

src/main/java/io/cryostat/discovery/KubeEndpointSlicesDiscovery.java

@@ -60,6 +60,8 @@
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.enterprise.event.Observes;
 import jakarta.inject.Inject;
+import jakarta.persistence.EntityManager;
+import jakarta.persistence.LockModeType;
 import jakarta.transaction.Transactional;
 import jakarta.transaction.Transactional.TxType;
 import org.apache.commons.lang3.StringUtils;
@@ -115,6 +117,8 @@ public class KubeEndpointSlicesDiscovery implements ResourceEventHandler<Endpoin
 
     @Inject EventBus bus;
 
+    @Inject EntityManager entityManager;
+
     @ConfigProperty(name = "cryostat.discovery.kubernetes.enabled")
     boolean enabled;
 
@@ -383,11 +387,18 @@ public void handleQueryEvent(NamespaceQueryEvent evt) {
     public void handleEndpointEvent(EndpointDiscoveryEvent evt) {
         String namespace = evt.namespace;
         DiscoveryNode realm = DiscoveryNode.getRealm(REALM).orElseThrow();
+        realm = entityManager.find(DiscoveryNode.class, realm.id, LockModeType.PESSIMISTIC_WRITE);
+        DiscoveryNode lockedRealm = realm;
         DiscoveryNode nsNode =
-                DiscoveryNode.getChild(realm, n -> n.name.equals(namespace))
-                        .orElse(
-                                DiscoveryNode.environment(
-                                        namespace, KubeDiscoveryNodeType.NAMESPACE));
+                DiscoveryNode.getChild(lockedRealm, n -> n.name.equals(namespace))
+                        .orElseGet(
+                                () -> {
+                                    DiscoveryNode created =
+                                            DiscoveryNode.environment(
+                                                    namespace, KubeDiscoveryNodeType.NAMESPACE);
+                                    created.parent = lockedRealm;
+                                    return created;
+                                });
 
         if (evt.eventKind == EventKind.FOUND) {
             persistOwnerChain(nsNode, evt.target, evt.objRef);
@@ -402,6 +413,7 @@ public void handleEndpointEvent(EndpointDiscoveryEvent evt) {
             realm.children.add(nsNode);
             nsNode.parent = realm;
         }
+        entityManager.flush();
         realm.persist();
     }
 
@@ -718,6 +730,7 @@ private void pruneOwnerChain(DiscoveryNode nsNode, Target target) {
             child = parent;
         }
 
+        entityManager.flush();
         nsNode.persist();
         target.delete();
     }
@@ -807,6 +820,7 @@ private void persistNodeChain(DiscoveryNode nsNode, Target target, DiscoveryNode
         }
 
         // Finally persist the namespace node
+        entityManager.flush();
         nsNode.persist();
     }
 

src/main/java/io/cryostat/discovery/PluginCallbackFactory.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright The Cryostat Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.cryostat.discovery;
+
+import java.net.URISyntaxException;
+
+import io.cryostat.discovery.DiscoveryPlugin.PluginCallback;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+/**
+ * Factory for creating PluginCallback instances. This abstraction allows tests to inject mock
+ * callbacks without making real network connections.
+ */
+@ApplicationScoped
+public class PluginCallbackFactory {
+
+    public PluginCallback create(DiscoveryPlugin plugin) throws URISyntaxException {
+        return PluginCallback.create(plugin);
+    }
+}