Add support for endpoint and server_name connection error labels (#138533)

Adds a protected RemoteConnectionStrategy method for adding
strategy specific attributes to connection error metric records.
This supports the monitoring dashboard so we can include the
endpoint and server_name information that will correspond with
the resolved address and error stack traces in the component
logs.

Relates: ES-12696

Author: JeremyDahlgren

server/src/main/java/org/elasticsearch/transport/ProxyConnectionStrategy.java

@@ -107,7 +107,7 @@ protected boolean shouldOpenMoreConnections() {
 
     @Override
     protected boolean strategyMustBeRebuilt(LinkedProjectConfig config) {
-        assert config instanceof ProxyLinkedProjectConfig : "expected config to be of type " + ProxyConnectionStrategy.class;
+        assert config instanceof ProxyLinkedProjectConfig : "expected config to be of type " + ProxyLinkedProjectConfig.class;
         final var proxyConfig = (ProxyLinkedProjectConfig) config;
         return proxyConfig.maxNumConnections() != maxNumConnections
             || configuredAddress.equals(proxyConfig.proxyAddress()) == false
@@ -119,6 +119,14 @@ protected ConnectionStrategy strategyType() {
         return ConnectionStrategy.PROXY;
     }
 
+    @Override
+    protected void addStrategySpecificConnectionErrorMetricAttributes(Map<String, Object> attributesMap) {
+        attributesMap.put("endpoint", configuredAddress);
+        if (configuredServerName != null) {
+            attributesMap.put("server_name", configuredServerName);
+        }
+    }
+
     @Override
     protected void connectImpl(ActionListener<Void> listener) {
         performProxyConnectionProcess(listener);

server/src/main/java/org/elasticsearch/transport/RemoteConnectionStrategy.java

@@ -31,6 +31,7 @@
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -237,19 +238,13 @@ private void connectionAttemptCompleted(@Nullable Exception e) {
         } else {
             logger.warn(msgSupplier, e);
             if (connectionAttemptFailures != null) {
-                connectionAttemptFailures.incrementBy(
-                    1,
-                    Map.of(
-                        "linked_project_id",
-                        linkedProjectId.toString(),
-                        "linked_project_alias",
-                        clusterAlias,
-                        "attempt",
-                        (isInitialAttempt ? ConnectionAttempt.initial : ConnectionAttempt.reconnect).toString(),
-                        "strategy",
-                        strategyType().toString()
-                    )
-                );
+                final var attributesMap = new HashMap<String, Object>();
+                attributesMap.put("linked_project_id", linkedProjectId.toString());
+                attributesMap.put("linked_project_alias", clusterAlias);
+                attributesMap.put("attempt", (isInitialAttempt ? ConnectionAttempt.initial : ConnectionAttempt.reconnect).toString());
+                attributesMap.put("strategy", strategyType().toString());
+                addStrategySpecificConnectionErrorMetricAttributes(attributesMap);
+                connectionAttemptFailures.incrementBy(1, attributesMap);
             }
         }
     }
@@ -264,6 +259,11 @@ boolean shouldRebuildConnection(LinkedProjectConfig config) {
 
     protected abstract ConnectionStrategy strategyType();
 
+    /**
+     * Add strategy-specific attributes for a new connection error metric record.  The default implementation is a no-op.
+     */
+    protected void addStrategySpecificConnectionErrorMetricAttributes(Map<String, Object> attributesMap) {}
+
     @Override
     public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
         if (shouldOpenMoreConnections()) {

server/src/test/java/org/elasticsearch/transport/ProxyConnectionStrategyTests.java

@@ -25,6 +25,8 @@
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.IndexVersions;
+import org.elasticsearch.telemetry.InstrumentType;
+import org.elasticsearch.telemetry.RecordingMeterRegistry;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.TransportVersionUtils;
 import org.elasticsearch.test.junit.annotations.TestLogging;
@@ -360,6 +362,44 @@ public void testConnectFailsWithNonRetryableException() {
         }
     }
 
+    public void testStrategySpecificConnectionErrorMetricAttributesAreAdded() {
+        try (
+            MockTransportService transport1 = startTransport("remote", VersionInformation.CURRENT, TransportVersion.current());
+            MockTransportService localService = MockTransportService.createNewService(
+                Settings.EMPTY,
+                VersionInformation.CURRENT,
+                TransportVersion.current(),
+                threadPool
+            )
+        ) {
+            final var address1 = transport1.boundAddress().publishAddress();
+            localService.addSendBehavior(address1, (connection, requestId, action, request, options) -> {
+                throw new ElasticsearchException("non-retryable");
+            });
+            localService.start();
+            localService.acceptIncomingRequests();
+
+            final var cfg = proxyStrategyConfig(clusterAlias, 1, address1.toString(), "address1_server_name");
+            final var connectFuture = new PlainActionFuture<RemoteClusterService.RemoteClusterConnectionStatus>();
+            localService.getRemoteClusterService().updateRemoteCluster(cfg, false, connectFuture);
+            final var exception = expectThrows(ElasticsearchException.class, connectFuture::actionGet);
+            assertThat(exception.getMessage(), containsString("non-retryable"));
+
+            assert localService.getTelemetryProvider() != null;
+            final var meterRegistry = localService.getTelemetryProvider().getMeterRegistry();
+            assert meterRegistry instanceof RecordingMeterRegistry;
+            final var metricRecorder = ((RecordingMeterRegistry) meterRegistry).getRecorder();
+            metricRecorder.collect();
+            final var counterName = RemoteClusterService.CONNECTION_ATTEMPT_FAILURES_COUNTER_NAME;
+            final var measurements = metricRecorder.getMeasurements(InstrumentType.LONG_COUNTER, counterName);
+            assertFalse(measurements.isEmpty());
+            final var measurement = measurements.getLast();
+            assertThat(measurement.getLong(), equalTo(1L));
+            assertThat(measurement.attributes().get("endpoint"), equalTo(address1.toString()));
+            assertThat(measurement.attributes().get("server_name"), equalTo("address1_server_name"));
+        }
+    }
+
     public void testClusterNameValidationPreventConnectingToDifferentClusters() throws Exception {
         Settings otherSettings = Settings.builder().put("cluster.name", "otherCluster").build();
 
@@ -600,13 +640,16 @@ public void testProxyStrategyWillNeedToBeRebuiltIfNumOfSocketsOrAddressesOrServe
                     assertFalse(strategy.shouldRebuildConnection(RemoteClusterSettings.toConfig(clusterAlias, noChange)));
                     Settings addressesChanged = Settings.builder()
                         .put(modeSetting.getKey(), "proxy")
-                        .put(addressesSetting.getKey(), remoteAddress.toString())
+                        .put(addressesSetting.getKey(), "foobar:8080")
+                        .put(socketConnections.getKey(), numOfConnections)
+                        .put(serverName.getKey(), "server-name")
                         .build();
                     assertTrue(strategy.shouldRebuildConnection(RemoteClusterSettings.toConfig(clusterAlias, addressesChanged)));
                     Settings socketsChanged = Settings.builder()
                         .put(modeSetting.getKey(), "proxy")
                         .put(addressesSetting.getKey(), remoteAddress.toString())
                         .put(socketConnections.getKey(), numOfConnections + 1)
+                        .put(serverName.getKey(), "server-name")
                         .build();
                     assertTrue(strategy.shouldRebuildConnection(RemoteClusterSettings.toConfig(clusterAlias, socketsChanged)));
                     Settings serverNameChange = Settings.builder()

server/src/test/java/org/elasticsearch/transport/RemoteConnectionStrategyTests.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 
+import java.util.Map;
 import java.util.Set;
 
 import static org.elasticsearch.test.MockLog.assertThatLogger;
@@ -267,7 +268,13 @@ public void testConnectionAttemptMetricsAndLogging() {
                         final var measurement = measurements.getLast();
                         assertThat(measurement.getLong(), equalTo(1L));
                         final var attributes = measurement.attributes();
-                        final var keySet = Set.of("linked_project_id", "linked_project_alias", "attempt", "strategy");
+                        final var keySet = Set.of(
+                            "linked_project_id",
+                            "linked_project_alias",
+                            "attempt",
+                            "strategy",
+                            "fake_metric_attribute_name"
+                        );
                         final var expectedAttemptType = isInitialConnectAttempt
                             ? RemoteConnectionStrategy.ConnectionAttempt.initial
                             : RemoteConnectionStrategy.ConnectionAttempt.reconnect;
@@ -276,6 +283,7 @@ public void testConnectionAttemptMetricsAndLogging() {
                         assertThat(attributes.get("linked_project_alias"), equalTo(alias));
                         assertThat(attributes.get("attempt"), equalTo(expectedAttemptType.toString()));
                         assertThat(attributes.get("strategy"), equalTo(strategy.strategyType().toString()));
+                        assertThat(attributes.get("fake_metric_attribute_name"), equalTo("fake_metric_attribute_value"));
                     }
                 }
             }
@@ -390,5 +398,10 @@ protected void connectImpl(ActionListener<Void> listener) {
         protected RemoteConnectionInfo.ModeInfo getModeInfo() {
             return null;
         }
+
+        @Override
+        protected void addStrategySpecificConnectionErrorMetricAttributes(Map<String, Object> attributeMap) {
+            attributeMap.put("fake_metric_attribute_name", "fake_metric_attribute_value");
+        }
     }
 }