Introduce SilentlyFailingRevocationChecker

* This checker catches failures during revocation checking and logs them
  in a connection logger

Signed-off-by: Yannic Klem <yannic.klem@bosch.io>

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/amqp/AmqpClientActor.java

@@ -274,7 +274,8 @@ protected void allocateResourcesOnConnection(final ClientConnected clientConnect
             jmsConnection.addConnectionListener(connectionListener);
             jmsSession = c.session;
         } else {
-            logger.info("ClientConnected was not JmsConnected as expected, ignoring as this probably was a reconnection");
+            logger.info(
+                    "ClientConnected was not JmsConnected as expected, ignoring as this probably was a reconnection");
         }
     }
 
@@ -419,7 +420,8 @@ private ActorRef startConnectionHandlingActor(final String suffix, final Connect
         final String namePrefix =
                 JMSConnectionHandlingActor.ACTOR_NAME_PREFIX + escapeActorName(connectionId() + "-" + suffix);
         final Props props =
-                JMSConnectionHandlingActor.propsWithOwnDispatcher(connection, this, jmsConnectionFactory);
+                JMSConnectionHandlingActor.propsWithOwnDispatcher(connection, this, jmsConnectionFactory,
+                        connectionLogger);
         return startChildActorConflictFree(namePrefix, props);
     }
 

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/amqp/ConnectionBasedJmsConnectionFactory.java

@@ -39,6 +39,7 @@
 import org.eclipse.ditto.model.connectivity.Connection;
 import org.eclipse.ditto.services.connectivity.messaging.config.Amqp10Config;
 import org.eclipse.ditto.services.connectivity.messaging.internal.ssl.SSLContextCreator;
+import org.eclipse.ditto.services.connectivity.messaging.monitoring.logs.ConnectionLogger;
 import org.slf4j.LoggerFactory;
 
 /**
@@ -70,8 +71,9 @@ public static ConnectionBasedJmsConnectionFactory getInstance(final Amqp10Config
     }
 
     @Override
-    public JmsConnection createConnection(final Connection connection, final ExceptionListener exceptionListener)
-            throws JMSException, NamingException {
+    public JmsConnection createConnection(final Connection connection, final ExceptionListener exceptionListener,
+            final ConnectionLogger connectionLogger) throws JMSException, NamingException {
+
         checkNotNull(connection, "Connection");
         checkNotNull(exceptionListener, "Exception Listener");
 
@@ -80,7 +82,8 @@ public JmsConnection createConnection(final Connection connection, final Excepti
                 (org.apache.qpid.jms.JmsConnectionFactory) ctx.lookup(connection.getId().toString());
 
         if (isSecuredConnection(connection) && connection.isValidateCertificates()) {
-            cf.setSslContext(SSLContextCreator.fromConnection(connection, null).withoutClientCertificate());
+            cf.setSslContext(SSLContextCreator.fromConnection(connection, null, connectionLogger)
+                    .withoutClientCertificate());
         }
 
         @SuppressWarnings("squid:S2095") final JmsConnection jmsConnection = (JmsConnection) cf.createConnection();

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/amqp/JMSConnectionHandlingActor.java

@@ -38,6 +38,7 @@
 import org.eclipse.ditto.services.connectivity.messaging.config.ConnectionConfig;
 import org.eclipse.ditto.services.connectivity.messaging.config.DittoConnectivityConfig;
 import org.eclipse.ditto.services.connectivity.messaging.internal.ImmutableConnectionFailure;
+import org.eclipse.ditto.services.connectivity.messaging.monitoring.logs.ConnectionLogger;
 import org.eclipse.ditto.services.utils.akka.LogUtil;
 import org.eclipse.ditto.services.utils.config.DefaultScopedConfig;
 import org.eclipse.ditto.signals.commands.connectivity.exceptions.ConnectionFailedException;
@@ -87,16 +88,18 @@ public final class JMSConnectionHandlingActor extends AbstractActor {
     private final Connection connection;
     private final ExceptionListener exceptionListener;
     private final JmsConnectionFactory jmsConnectionFactory;
+    private final ConnectionLogger connectionLogger;
 
     @Nullable private Session currentSession = null;
 
     @SuppressWarnings("unused")
     private JMSConnectionHandlingActor(final Connection connection, final ExceptionListener exceptionListener,
-            final JmsConnectionFactory jmsConnectionFactory) {
+            final JmsConnectionFactory jmsConnectionFactory, final ConnectionLogger connectionLogger) {
 
         this.connection = checkNotNull(connection, "connection");
         this.exceptionListener = exceptionListener;
         this.jmsConnectionFactory = jmsConnectionFactory;
+        this.connectionLogger = connectionLogger;
     }
 
     /**
@@ -105,17 +108,19 @@ private JMSConnectionHandlingActor(final Connection connection, final ExceptionL
      * @param connection the connection
      * @param exceptionListener the exception listener
      * @param jmsConnectionFactory the jms connection factory
+     * @param connectionLogger used to log failures during certificate validation.
      * @return the Akka configuration Props object.
      */
     static Props props(final Connection connection, final ExceptionListener exceptionListener,
-            final JmsConnectionFactory jmsConnectionFactory) {
+            final JmsConnectionFactory jmsConnectionFactory, final ConnectionLogger connectionLogger) {
 
-        return Props.create(JMSConnectionHandlingActor.class, connection, exceptionListener, jmsConnectionFactory);
+        return Props.create(JMSConnectionHandlingActor.class, connection, exceptionListener, jmsConnectionFactory,
+                connectionLogger);
     }
 
     static Props propsWithOwnDispatcher(final Connection connection, final ExceptionListener exceptionListener,
-            final JmsConnectionFactory jmsConnectionFactory) {
-        return props(connection, exceptionListener, jmsConnectionFactory)
+            final JmsConnectionFactory jmsConnectionFactory, final ConnectionLogger connectionLogger) {
+        return props(connection, exceptionListener, jmsConnectionFactory, connectionLogger)
                 .withDispatcher(DISPATCHER_NAME);
     }
 
@@ -362,7 +367,7 @@ private JmsConnection createJmsConnection() {
                         ConnectionBasedJmsConnectionFactory
                                 .buildAmqpConnectionUriFromConnection(connection, amqp10Config));
             }
-            return jmsConnectionFactory.createConnection(connection, exceptionListener);
+            return jmsConnectionFactory.createConnection(connection, exceptionListener, connectionLogger);
         });
     }
 

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/amqp/JmsConnectionFactory.java

@@ -18,6 +18,7 @@
 
 import org.apache.qpid.jms.JmsConnection;
 import org.eclipse.ditto.model.connectivity.Connection;
+import org.eclipse.ditto.services.connectivity.messaging.monitoring.logs.ConnectionLogger;
 
 /**
  * Creates a new {@link javax.jms.Connection}.
@@ -29,11 +30,12 @@ public interface JmsConnectionFactory {
      *
      * @param connection the connection to use for the returned JMS Connection.
      * @param exceptionListener the ExceptionListener to configure for the returned JMS Connection.
+     * @param connectionLogger used to log failures during certificate validation.
      * @return the JMS Connection.
      * @throws javax.jms.JMSException if the context could not be created.
      * @throws javax.naming.NamingException if the identifier of {@code connection} could not be found in the Context.
      */
-    JmsConnection createConnection(Connection connection, ExceptionListener exceptionListener)
-            throws JMSException, NamingException;
+    JmsConnection createConnection(Connection connection, ExceptionListener exceptionListener,
+            ConnectionLogger connectionLogger) throws JMSException, NamingException;
 
 }

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/httppush/DefaultHttpPushFactory.java

@@ -24,6 +24,7 @@
 import org.eclipse.ditto.model.connectivity.ConnectionId;
 import org.eclipse.ditto.services.connectivity.messaging.config.HttpPushConfig;
 import org.eclipse.ditto.services.connectivity.messaging.internal.ssl.SSLContextCreator;
+import org.eclipse.ditto.services.connectivity.messaging.monitoring.logs.ConnectionLogger;
 
 import akka.actor.ActorSystem;
 import akka.event.LoggingAdapter;
@@ -74,15 +75,16 @@ private DefaultHttpPushFactory(final ConnectionId connectionId, final Uri baseUr
         this.httpsConnectionContext = httpsConnectionContext;
     }
 
-    static HttpPushFactory of(final Connection connection, final HttpPushConfig httpPushConfig) {
+    static HttpPushFactory of(final Connection connection, final HttpPushConfig httpPushConfig,
+            final ConnectionLogger connectionLogger) {
         final ConnectionId connectionId = connection.getId();
         final Uri baseUri = Uri.create(connection.getUri());
         final int parallelism = parseParallelism(connection.getSpecificConfig());
 
         final HttpsConnectionContext httpsConnectionContext;
         if (HttpPushValidator.isSecureScheme(baseUri.getScheme())) {
             final SSLContextCreator sslContextCreator =
-                    SSLContextCreator.fromConnection(connection, DittoHeaders.empty());
+                    SSLContextCreator.fromConnection(connection, DittoHeaders.empty(), connectionLogger);
             final SSLContext sslContext = connection.getCredentials()
                     .map(credentials -> credentials.accept(sslContextCreator))
                     .orElse(sslContextCreator.withoutClientCertificate());

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/httppush/HttpPublisherActor.java

@@ -150,10 +150,8 @@ static Props props(final Connection connection, final HttpPushFactory factory) {
     }
 
     private ConnectionLogger getConnectionLogger(final Connection connection) {
-        final MonitoringConfig monitoringConfig = connectivityConfig.getMonitoringConfig();
-        final MonitoringLoggerConfig loggerConfig = monitoringConfig.logger();
-        final ConnectionLoggerRegistry connectionLoggerRegistry = ConnectionLoggerRegistry.fromConfig(loggerConfig);
-        return connectionLoggerRegistry.forConnection(connection.getId());
+        final MonitoringLoggerConfig loggerConfig = connectivityConfig.getMonitoringConfig().logger();
+        return ConnectionLogger.getInstance(connection.getId(), loggerConfig);
     }
 
     @Override
@@ -322,7 +320,8 @@ private CompletionStage<CommandResponse<?>> toCommandResponseOrAcknowledgement(f
                 if (DittoAcknowledgementLabel.LIVE_RESPONSE.equals(label)) {
                     // Live-Response is declared as issued ack => parse live response from response
                     if (isMessageCommand) {
-                        result = toMessageCommandResponse((MessageCommand<?, ?>) signal, dittoHeaders, body, statusCode);
+                        result =
+                                toMessageCommandResponse((MessageCommand<?, ?>) signal, dittoHeaders, body, statusCode);
                     } else {
                         result = null;
                     }

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/httppush/HttpPushClientActor.java

@@ -35,9 +35,11 @@
 import org.eclipse.ditto.services.connectivity.messaging.BaseClientActor;
 import org.eclipse.ditto.services.connectivity.messaging.config.DittoConnectivityConfig;
 import org.eclipse.ditto.services.connectivity.messaging.config.HttpPushConfig;
+import org.eclipse.ditto.services.connectivity.messaging.config.MonitoringLoggerConfig;
 import org.eclipse.ditto.services.connectivity.messaging.internal.ClientConnected;
 import org.eclipse.ditto.services.connectivity.messaging.internal.ClientDisconnected;
 import org.eclipse.ditto.services.connectivity.messaging.internal.ssl.SSLContextCreator;
+import org.eclipse.ditto.services.connectivity.messaging.monitoring.logs.ConnectionLogger;
 import org.eclipse.ditto.services.utils.config.DefaultScopedConfig;
 import org.eclipse.ditto.signals.commands.connectivity.modify.TestConnection;
 
@@ -54,6 +56,7 @@ public final class HttpPushClientActor extends BaseClientActor {
     private static final int PROXY_CONNECT_TIMEOUT_SECONDS = 15;
 
     private final HttpPushFactory factory;
+    private final ConnectionLogger connectionLogger;
 
     @Nullable
     private ActorRef httpPublisherActor;
@@ -63,12 +66,15 @@ public final class HttpPushClientActor extends BaseClientActor {
     private HttpPushClientActor(final Connection connection, final ActorRef connectionActor) {
         super(connection, ActorRef.noSender(), connectionActor);
 
-        httpPushConfig = DittoConnectivityConfig.of(
+        final DittoConnectivityConfig connectivityConfig = DittoConnectivityConfig.of(
                 DefaultScopedConfig.dittoScoped(getContext().getSystem().settings().config())
-        )
+        );
+        httpPushConfig = connectivityConfig
                 .getConnectionConfig()
                 .getHttpPushConfig();
-        factory = HttpPushFactory.of(connection, httpPushConfig);
+        final MonitoringLoggerConfig loggerConfig = connectivityConfig.getMonitoringConfig().logger();
+        connectionLogger = ConnectionLogger.getInstance(connection.getId(), loggerConfig);
+        factory = HttpPushFactory.of(connection, httpPushConfig, connectionLogger);
     }
 
     /**
@@ -153,7 +159,7 @@ private CompletionStage<Status.Status> testSSL(final Connection connection, fina
         } else {
             // check without HTTP proxy
             final SSLContextCreator sslContextCreator =
-                    SSLContextCreator.fromConnection(connection, DittoHeaders.empty());
+                    SSLContextCreator.fromConnection(connection, DittoHeaders.empty(), connectionLogger);
             final SSLSocketFactory socketFactory = connection.getCredentials()
                     .map(credentials -> credentials.accept(sslContextCreator))
                     .orElse(sslContextCreator.withoutClientCertificate()).getSocketFactory();

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/httppush/HttpPushFactory.java

@@ -14,6 +14,7 @@
 
 import org.eclipse.ditto.model.connectivity.Connection;
 import org.eclipse.ditto.services.connectivity.messaging.config.HttpPushConfig;
+import org.eclipse.ditto.services.connectivity.messaging.monitoring.logs.ConnectionLogger;
 
 import akka.actor.ActorSystem;
 import akka.event.LoggingAdapter;
@@ -60,7 +61,8 @@ public interface HttpPushFactory {
      * @param httpPushConfig configuration of Http connections.
      * @return the HTTP-push-factory.
      */
-    static HttpPushFactory of(final Connection connection, final HttpPushConfig httpPushConfig) {
-        return DefaultHttpPushFactory.of(connection, httpPushConfig);
+    static HttpPushFactory of(final Connection connection, final HttpPushConfig httpPushConfig,
+            final ConnectionLogger connectionLogger) {
+        return DefaultHttpPushFactory.of(connection, httpPushConfig, connectionLogger);
     }
 }

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/internal/ssl/DittoTrustManagerFactory.java

@@ -22,23 +22,28 @@
 import javax.net.ssl.TrustManagerFactory;
 import javax.net.ssl.TrustManagerFactorySpi;
 
+import org.eclipse.ditto.model.base.headers.DittoHeaders;
 import org.eclipse.ditto.model.connectivity.Connection;
+import org.eclipse.ditto.services.connectivity.messaging.monitoring.logs.ConnectionLogger;
 
 /**
  * Simple wrapper around {@link TrustManagerFactory} that wraps the returned {@link TrustManager}s in
  * {@link DittoTrustManager}s and delegates all other invocations.
  */
 public final class DittoTrustManagerFactory extends TrustManagerFactory {
 
-    private static final TrustManagerFactoryFactory FACTORY = TrustManagerFactoryFactory.getInstance();
+    private static final TrustManagerFactoryFactory FACTORY =
+            TrustManagerFactoryFactory.getInstance(DittoHeaders.empty());
 
-    public static DittoTrustManagerFactory from(final Connection connection) {
+    public static DittoTrustManagerFactory from(final Connection connection, final ConnectionLogger connectionLogger) {
         final String hostname = connection.getHostname();
-        return new DittoTrustManagerFactory(FACTORY.newTrustManagerFactory(connection), hostname);
+        return new DittoTrustManagerFactory(FACTORY.newTrustManagerFactory(connection, connectionLogger), hostname);
     }
 
-    static DittoTrustManagerFactory from(@Nullable final String trustedCertificates, final String hostname) {
-        return new DittoTrustManagerFactory(FACTORY.newTrustManagerFactory(trustedCertificates), hostname);
+    static DittoTrustManagerFactory from(@Nullable final String trustedCertificates, final String hostname,
+            final ConnectionLogger connectionLogger) {
+        return new DittoTrustManagerFactory(FACTORY.newTrustManagerFactory(trustedCertificates, connectionLogger),
+                hostname);
     }
 
     private DittoTrustManagerFactory(final TrustManagerFactory delegate, final String hostname) {
@@ -60,4 +65,4 @@ protected TrustManager[] engineGetTrustManagers() {
             }
         }, delegate.getProvider(), delegate.getAlgorithm());
     }
-}
+}