[#1843] [AMQ-9687] Add metrics about error and reconnect counts to network co…

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java

@@ -99,6 +99,21 @@ public void resetStats(){
         }
     }
 
+    @Override
+    public long getStartedTimestamp() {
+        return bridge.getStartedTimestamp();
+    }
+
+    @Override
+    public long getLocalExceptionCount() {
+        return bridge.getLocalExceptionCount();
+    }
+
+    @Override
+    public long getRemoteExceptionCount() {
+        return bridge.getRemoteExceptionCount();
+    }
+
     public void addNetworkDestinationView(NetworkDestinationView networkDestinationView){
         networkDestinationViewList.add(networkDestinationView);
     }

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java

@@ -40,4 +40,9 @@ public interface NetworkBridgeViewMBean extends Service {
 
     void resetStats();
 
+    long getStartedTimestamp();
+
+    long getLocalExceptionCount();
+
+    long getRemoteExceptionCount();
 }

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorView.java

@@ -219,4 +219,29 @@ public void setRemoteUserName(String remoteUserName) {
     public boolean isAutoStart() {
         return connector.isAutoStart();
     }
+
+    @Override
+    public long getStartedTimestamp() {
+        return connector.getStartedTimestamp();
+    }
+
+    @Override
+    public long getStoppedTimestamp() {
+        return connector.getStoppedTimestamp();
+    }
+
+    @Override
+    public long getBridgeExceptionCount() {
+        return connector.getBridgeExceptionCounter().getCount();
+    }
+
+    @Override
+    public long getLocalExceptionCount() {
+        return connector.getLocalExceptionCounter().getCount();
+    }
+
+    @Override
+    public long getRemoteExceptionCount() {
+        return connector.getRemoteExceptionCounter().getCount();
+    }
 }

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorViewMBean.java

@@ -94,4 +94,14 @@ public interface NetworkConnectorViewMBean extends Service {
     void setRemotePassword(String remotePassword);
 
     boolean isAutoStart();
+
+    long getStartedTimestamp();
+
+    long getStoppedTimestamp();
+
+    long getBridgeExceptionCount();
+
+    long getLocalExceptionCount();
+
+    long getRemoteExceptionCount();
 }

activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java

@@ -37,6 +37,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Pattern;
 
 import javax.management.ObjectName;
@@ -129,6 +130,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
     protected final AtomicBoolean localBridgeStarted = new AtomicBoolean(false);
     protected final AtomicBoolean remoteBridgeStarted = new AtomicBoolean(false);
     protected final AtomicBoolean bridgeFailed = new AtomicBoolean();
+    // Ensures a single bridge failure counts exactly one local/remote exception
+    protected final AtomicBoolean bridgeExceptionCounted = new AtomicBoolean();
     protected final AtomicBoolean disposed = new AtomicBoolean();
     protected BrokerId localBrokerId;
     protected ActiveMQDestination[] excludedDestinations;
@@ -171,6 +174,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
     private final ExecutorService syncExecutor = Executors.newSingleThreadExecutor();
     private Transport duplexInboundLocalBroker = null;
     private ProducerInfo duplexInboundLocalProducerInfo;
+    private final AtomicLong startedTimestamp = new AtomicLong(0L);
 
     public DemandForwardingBridgeSupport(NetworkBridgeConfiguration configuration, Transport localBroker, Transport remoteBroker) {
         this.configuration = configuration;
@@ -194,7 +198,15 @@ public void start() throws Exception {
                 throw new IllegalArgumentException("BrokerService is null on " + this);
             }
 
-            networkBridgeStatistics.setEnabled(brokerService.isEnableStatistics());
+            if (brokerService.isEnableStatistics()) {
+                networkBridgeStatistics.setEnabled(true);
+                // Bridges are ephemeral - parent this bridge's statistics to the owning
+                // connector so its counts (including local/remote exceptions) roll up
+                // automatically and survive the bridge being torn down and recreated.
+                if (configuration instanceof NetworkConnector) {
+                    networkBridgeStatistics.setParent(((NetworkConnector) configuration).getNetworkBridgeStatistics());
+                }
+            }
 
             if (isDuplex()) {
                 duplexInboundLocalBroker = NetworkBridgeFactory.createLocalAsyncTransport(brokerService.getBroker().getVmConnectorURI());
@@ -260,6 +272,8 @@ public void onException(IOException error) {
                     triggerStartAsyncNetworkBridgeCreation();
                 } catch (IOException e) {
                     LOG.warn("Caught exception from remote start", e);
+                } finally {
+                    startedTimestamp.set(System.currentTimeMillis());
                 }
             } else {
                 LOG.warn("Bridge was disposed before the start() method was fully executed.");
@@ -336,6 +350,7 @@ public void run() {
                     startedLatch.countDown();
                     localStartedLatch.countDown();
                     staticDestinationsLatch.countDown();
+                    startedTimestamp.set(0L);
 
                     ss.throwFirstException();
                 }
@@ -650,12 +665,21 @@ protected void startRemoteBridge() throws Exception {
     @Override
     public void serviceRemoteException(Throwable error) {
         if (!disposed.get()) {
+            // Count the single genuine failure that trips disposal, not the peer exceptions
+            // fired during teardown nor a second callback racing in before disposal completes.
+            // This rolls up to the owning connector automatically via the parent statistics
+            // wired in start().
+            if (bridgeExceptionCounted.compareAndSet(false, true)) {
+                networkBridgeStatistics.getRemoteExceptionCount().increment();
+            }
+
             if (error instanceof SecurityException || error instanceof GeneralSecurityException) {
                 LOG.error("Network connection between {} and {} shutdown due to a remote error: {}", localBroker, remoteBroker, error.toString());
             } else {
                 LOG.warn("Network connection between {} and {} shutdown due to a remote error: {}", localBroker, remoteBroker, error.toString());
             }
             LOG.debug("The remote Exception was: {}", error, error);
+
             brokerService.getTaskRunnerFactory().execute(new Runnable() {
                 @Override
                 public void run() {
@@ -1114,6 +1138,7 @@ public void serviceLocalException(Throwable error) {
 
     public void serviceLocalException(MessageDispatch messageDispatch, Throwable error) {
         LOG.trace("serviceLocalException: disposed {} ex", disposed.get(), error);
+
         if (!disposed.get()) {
             if (error instanceof DestinationDoesNotExistException && ((DestinationDoesNotExistException) error).isTemporary()) {
                 // not a reason to terminate the bridge - temps can disappear with
@@ -1134,6 +1159,14 @@ public void serviceLocalException(MessageDispatch messageDispatch, Throwable err
                 return;
             }
 
+            // Count the single genuine failure that trips disposal, not the peer exceptions
+            // fired during teardown, a second callback racing in before disposal completes,
+            // nor the ignorable temporary-destination errors handled above. This rolls up to
+            // the owning connector automatically via the parent statistics wired in start().
+            if (bridgeExceptionCounted.compareAndSet(false, true)) {
+                networkBridgeStatistics.getLocalExceptionCount().increment();
+            }
+
             LOG.info("Network connection between {} and {} shutdown due to a local error: {}", localBroker, remoteBroker, error);
             LOG.debug("The local Exception was: {}", error, error);
 
@@ -1920,6 +1953,21 @@ public long getEnqueueCounter() {
         return networkBridgeStatistics.getEnqueues().getCount();
     }
 
+    @Override
+    public long getStartedTimestamp() {
+        return startedTimestamp.get();
+    }
+
+    @Override
+    public long getLocalExceptionCount() {
+        return networkBridgeStatistics.getLocalExceptionCount().getCount();
+    }
+
+    @Override
+    public long getRemoteExceptionCount() {
+        return networkBridgeStatistics.getRemoteExceptionCount().getCount();
+    }
+
     @Override
     public NetworkBridgeStatistics getNetworkBridgeStatistics() {
         return networkBridgeStatistics;

activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java

@@ -131,6 +131,7 @@ public void onServiceAdd(DiscoveryEvent event) {
                 try {
                     remoteTransport = TransportFactory.connect(connectUri);
                 } catch (Exception e) {
+                    networkBridgeStatistics.getRemoteExceptionCount().increment();
                     LOG.warn("Could not connect to remote URI: {}: {}", connectUri, e.getMessage());
                     LOG.debug("Connection failure exception: ", e);
                     try {
@@ -143,6 +144,7 @@ public void onServiceAdd(DiscoveryEvent event) {
                 try {
                     localTransport = createLocalTransport();
                 } catch (Exception e) {
+                    networkBridgeStatistics.getLocalExceptionCount().increment();
                     ServiceSupport.dispose(remoteTransport);
                     LOG.warn("Could not connect to local URI: {}: {}", localURI, e.getMessage());
                     LOG.debug("Connection failure exception: ", e);
@@ -164,6 +166,7 @@ public void onServiceAdd(DiscoveryEvent event) {
                 }
                 bridge.start();
             } catch (Exception e) {
+                bridgeExceptionCounter.increment();
                 ServiceSupport.dispose(localTransport);
                 ServiceSupport.dispose(remoteTransport);
                 LOG.warn("Could not start network bridge between: {} and: {} due to: {}", localURI, uri, e.getMessage());

activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java

@@ -95,4 +95,10 @@ public interface NetworkBridge extends Service {
     ObjectName getMbeanObjectName();
 
     void resetStats();
+
+    long getStartedTimestamp();
+
+    long getLocalExceptionCount();
+
+    long getRemoteExceptionCount();
 }

activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java

@@ -16,6 +16,8 @@
  */
 package org.apache.activemq.network;
 
+import java.util.Set;
+
 import org.apache.activemq.management.CountStatisticImpl;
 import org.apache.activemq.management.StatsImpl;
 
@@ -29,13 +31,17 @@ public class NetworkBridgeStatistics extends StatsImpl {
     protected CountStatisticImpl enqueues;
     protected CountStatisticImpl dequeues;
     protected CountStatisticImpl receivedCount;
+    protected CountStatisticImpl localExceptionCount;
+    protected CountStatisticImpl remoteExceptionCount;
 
     public NetworkBridgeStatistics() {
         enqueues = new CountStatisticImpl("enqueues", "The current number of enqueues this bridge has, which is the number of potential messages to be forwarded.");
         dequeues = new CountStatisticImpl("dequeues", "The current number of dequeues this bridge has, which is the number of messages received by the remote broker.");
         receivedCount = new CountStatisticImpl("receivedCount", "The number of messages that have been received by the NetworkBridge from the remote broker.  Only applies for Duplex bridges.");
+        localExceptionCount = new CountStatisticImpl("localExceptionCount", "The number of exceptions that have been received by the NetworkBridge from the local broker.");
+        remoteExceptionCount = new CountStatisticImpl("remoteExceptionCount", "The number of exceptions that have been received by the NetworkBridge from the remote broker.");
 
-        addStatistics(Set.of(enqueues, dequeues, receivedCount));
+        addStatistics(Set.of(enqueues, dequeues, receivedCount, localExceptionCount, remoteExceptionCount));
     }
 
     /**
@@ -68,13 +74,35 @@ public CountStatisticImpl getReceivedCount() {
         return receivedCount;
     }
 
+    /**
+     * The current number of exceptions this bridge has, which is the number of
+     * exceptions received from the remote broker.
+     *
+     * @return
+     */
+    public CountStatisticImpl getLocalExceptionCount() {
+        return localExceptionCount;
+    }
+
+    /**
+     * The current number of exceptions this bridge has, which is the number of
+     * exceptions received from the remote broker.
+     *
+     * @return
+     */
+    public CountStatisticImpl getRemoteExceptionCount() {
+        return remoteExceptionCount;
+    }
+
     @Override
     public void reset() {
         if (this.isDoReset()) {
             super.reset();
             enqueues.reset();
             dequeues.reset();
             receivedCount.reset();
+            localExceptionCount.reset();
+            remoteExceptionCount.reset();
         }
     }
 
@@ -84,17 +112,23 @@ public void setEnabled(boolean enabled) {
         enqueues.setEnabled(enabled);
         dequeues.setEnabled(enabled);
         receivedCount.setEnabled(enabled);
+        localExceptionCount.setEnabled(enabled);
+        remoteExceptionCount.setEnabled(enabled);
     }
 
     public void setParent(NetworkBridgeStatistics parent) {
         if (parent != null) {
             enqueues.setParent(parent.enqueues);
             dequeues.setParent(parent.dequeues);
             receivedCount.setParent(parent.receivedCount);
+            localExceptionCount.setParent(parent.localExceptionCount);
+            remoteExceptionCount.setParent(parent.remoteExceptionCount);
         } else {
             enqueues.setParent(null);
             dequeues.setParent(null);
             receivedCount.setParent(null);
+            localExceptionCount.setParent(null);
+            remoteExceptionCount.setParent(null);
         }
     }