Deprecate CloudstackSnitch and remove duplicate code in snitches

The patch also refactors existing cloud snitches to get rid of the duplicate code,
this is the logical follow-up of CASSANDRA-16555 where AbstractCloudMetadataServiceConnector was introduced.

patch by Stefan Miklosovic; reviewed by Jacek Lewandowski, Jackson Fleming and Maxwell Guo for CASSANDRA-18438

CHANGES.txt

@@ -1,4 +1,5 @@
 5.0
+ * Deprecate CloudstackSnitch and remove duplicate code in snitches (CASSANDRA-18438)
  * Add support for vectors in UDFs (CASSANDRA-18613)
  * Improve vector value validation errors (CASSANDRA-18652)
  * Upgrade Guava to 32.0.1 (CASSANDRA-18645)

NEWS.txt

@@ -223,6 +223,8 @@ Deprecation
       for partition tombstones. That guardrail is based on the properties `partition_tombstones_warn_threshold` and
       `partition_tombstones_fail_threshold`. The warn threshold has a very similar behaviour to the old config property.
       The old property is still supported for backward compatibility, but now it is disabled by default.
+    - CloudstackSnitch is marked as deprecated and it is not actively maintained anymore. It is scheduled to be removed 
+      in the next major version of Cassandra.
 
 4.1
 ===

conf/cassandra-rackdc.properties

@@ -19,8 +19,7 @@
 dc=dc1
 rack=rack1
 
-# Add a suffix to a datacenter name. Used by the Ec2Snitch and Ec2MultiRegionSnitch
-# to append a string to the EC2 region name.
+# Add a suffix to a datacenter name. Used by all cloud-based snitches.
 #dc_suffix=
 
 # Uncomment the following line to make this snitch prefer the internal ip when possible, as the Ec2MultiRegionSnitch does.
@@ -43,3 +42,15 @@ rack=rack1
 # If AWS IMDS of v2 is configured, ec2_metadata_token_ttl_seconds says how many seconds a token will be valid until
 # it is refreshed. Defaults to 21600. Can not be smaller than 30 and bigger than 21600. Has to be an integer.
 # ec2_metadata_token_ttl_seconds=21600
+
+# For all cloud-based snitches, there are following options available:
+#
+# Property to change metadata service url for a cloud-based snitch. The endpoint of a particular
+# snitch will be appended to this property. A user is not normally using this property, it is here
+# for tweaking the url of a service itself, e.g. for testing purposes.
+# metadata_url=http://some/service
+#
+# Sets a specified timeout value, in duration format, to be used when opening a communications link to metadata service,
+# referenced by an URLConnection. The timeout of zero (0s) is interpreted as an infinite timeout.
+# Defaults to 30 seconds.
+# metadata_request_timeout=30s

conf/cassandra.yaml

@@ -1298,6 +1298,17 @@ slow_query_log_timeout: 500ms
 #    Proximity is determined by rack and data center, which are
 #    explicitly configured in cassandra-topology.properties.
 #
+# AlibabaCloudSnitch:
+#    Snitch for getting dc and rack of a node from metadata service of Alibaba cloud.
+#    This snitch that assumes an ECS region is a DC and an ECS availability_zone is a rack.
+#
+# CloudstackSnitch:
+#    A snitch that assumes a Cloudstack Zone follows the typical convention 
+#    country-location-az and uses a country/location tuple as a datacenter
+#    and the availability zone as a rack.
+#    WARNING: This snitch is deprecated and it is scheduled to be removed 
+#    in the next major version of Cassandra.
+#
 # Ec2Snitch:
 #    Appropriate for EC2 deployments in a single Region. Loads Region
 #    and Availability Zone information from the EC2 API. The Region is
@@ -1313,6 +1324,10 @@ slow_query_log_timeout: 500ms
 #    traffic, Cassandra will switch to the private IP after
 #    establishing a connection.)
 #
+# GoogleCloudSnitch:
+#    Snitch for getting dc and rack of a node from metadata service of Google cloud.
+#    This snitch that assumes an GCE region is a DC and an GCE availability_zone is a rack.
+#
 # RackInferringSnitch:
 #    Proximity is determined by rack and data center, which are
 #    assumed to correspond to the 3rd and 2nd octet of each node's IP

src/java/org/apache/cassandra/locator/AbstractCloudMetadataServiceConnector.java

@@ -22,30 +22,72 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URISyntaxException;
 import java.net.URL;
 import java.util.Map;
 
 import com.google.common.collect.ImmutableMap;
 
+import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.config.DurationSpec;
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+import static java.lang.String.format;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 abstract class AbstractCloudMetadataServiceConnector
 {
+    static final String METADATA_URL_PROPERTY = "metadata_url";
+    static final String METADATA_REQUEST_TIMEOUT_PROPERTY = "metadata_request_timeout";
+    static final String DEFAULT_METADATA_REQUEST_TIMEOUT = "30s";
+
     protected final String metadataServiceUrl;
+    protected final int requestTimeoutMs;
 
-    protected AbstractCloudMetadataServiceConnector(String metadataServiceUrl)
+    public AbstractCloudMetadataServiceConnector(SnitchProperties properties)
     {
-        this.metadataServiceUrl = metadataServiceUrl;
+        String parsedMetadataServiceUrl = properties.get(METADATA_URL_PROPERTY, null);
+
+        try
+        {
+            URL url = new URL(parsedMetadataServiceUrl);
+            url.toURI();
+
+            this.metadataServiceUrl = parsedMetadataServiceUrl;
+        }
+        catch (MalformedURLException | IllegalArgumentException | URISyntaxException ex)
+        {
+            throw new ConfigurationException(format("Snitch metadata service URL '%s' is invalid. Please review snitch properties " +
+                                                    "defined in the configured '%s' configuration file.",
+                                                    parsedMetadataServiceUrl,
+                                                    CassandraRelevantProperties.CASSANDRA_RACKDC_PROPERTIES.getKey()),
+                                             ex);
+        }
+
+
+        String metadataRequestTimeout = properties.get(METADATA_REQUEST_TIMEOUT_PROPERTY, DEFAULT_METADATA_REQUEST_TIMEOUT);
+
+        try
+        {
+            this.requestTimeoutMs = new DurationSpec.IntMillisecondsBound(metadataRequestTimeout).toMilliseconds();
+        }
+        catch (IllegalArgumentException ex)
+        {
+            throw new ConfigurationException(format("%s as value of %s is invalid duration! " + ex.getMessage(),
+                                                    metadataRequestTimeout,
+                                                    METADATA_REQUEST_TIMEOUT_PROPERTY));
+        }
     }
 
-    public String apiCall(String query) throws IOException
+    public final String apiCall(String query) throws IOException
     {
-        return apiCall(metadataServiceUrl, query, 200);
+        return apiCall(metadataServiceUrl, query, "GET", ImmutableMap.of(), 200);
     }
 
-    public String apiCall(String url, String query, int expectedResponseCode) throws IOException
+    public final String apiCall(String query, Map<String, String> extraHeaders) throws IOException
     {
-        return apiCall(url, query, "GET", ImmutableMap.of(), expectedResponseCode);
+        return apiCall(metadataServiceUrl, query, "GET", extraHeaders, 200);
     }
 
     public String apiCall(String url,
@@ -61,6 +103,7 @@ public String apiCall(String url,
             conn = (HttpURLConnection) new URL(url + query).openConnection();
             extraHeaders.forEach(conn::setRequestProperty);
             conn.setRequestMethod(method);
+            conn.setConnectTimeout(requestTimeoutMs);
             if (conn.getResponseCode() != expectedResponseCode)
                 throw new HttpException(conn.getResponseCode(), conn.getResponseMessage());
 
@@ -84,6 +127,14 @@ public String apiCall(String url,
         }
     }
 
+    @Override
+    public String toString()
+    {
+        return format("%s{%s=%s,%s=%s}", getClass().getName(),
+                      METADATA_URL_PROPERTY, metadataServiceUrl,
+                      METADATA_REQUEST_TIMEOUT_PROPERTY, requestTimeoutMs);
+    }
+
     public static final class HttpException extends IOException
     {
         public final int responseCode;
@@ -96,4 +147,12 @@ public HttpException(int responseCode, String responseMessage)
             this.responseMessage = responseMessage;
         }
     }
+
+    public static class DefaultCloudMetadataServiceConnector extends AbstractCloudMetadataServiceConnector
+    {
+        public DefaultCloudMetadataServiceConnector(SnitchProperties properties)
+        {
+            super(properties);
+        }
+    }
 }

src/java/org/apache/cassandra/locator/AbstractCloudMetadataServiceSnitch.java

@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.cassandra.locator;
+
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.gms.ApplicationState;
+import org.apache.cassandra.gms.EndpointState;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Pair;
+
+import static java.lang.String.format;
+
+abstract class AbstractCloudMetadataServiceSnitch extends AbstractNetworkTopologySnitch
+{
+    static final Logger logger = LoggerFactory.getLogger(AbstractCloudMetadataServiceSnitch.class);
+
+    static final String DEFAULT_DC = "UNKNOWN-DC";
+    static final String DEFAULT_RACK = "UNKNOWN-RACK";
+
+    protected final AbstractCloudMetadataServiceConnector connector;
+    protected final SnitchProperties snitchProperties;
+
+    private final String localRack;
+    private final String localDc;
+
+    private Map<InetAddressAndPort, Map<String, String>> savedEndpoints;
+
+    public AbstractCloudMetadataServiceSnitch(AbstractCloudMetadataServiceConnector connector,
+                                              SnitchProperties snitchProperties,
+                                              Pair<String, String> dcAndRack)
+    {
+        this.connector = connector;
+        this.snitchProperties = snitchProperties;
+        this.localDc = dcAndRack.left;
+        this.localRack = dcAndRack.right;
+
+        logger.info(format("%s using datacenter: %s, rack: %s, connector: %s, properties: %s",
+                           getClass().getName(), getLocalDatacenter(), getLocalRack(), connector, snitchProperties));
+    }
+
+    @Override
+    public final String getLocalRack()
+    {
+        return localRack;
+    }
+
+    @Override
+    public final String getLocalDatacenter()
+    {
+        return localDc;
+    }
+
+    @Override
+    public final String getRack(InetAddressAndPort endpoint)
+    {
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
+            return getLocalRack();
+        EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
+        if (state == null || state.getApplicationState(ApplicationState.RACK) == null)
+        {
+            if (savedEndpoints == null)
+                savedEndpoints = SystemKeyspace.loadDcRackInfo();
+            if (savedEndpoints.containsKey(endpoint))
+                return savedEndpoints.get(endpoint).get("rack");
+            return DEFAULT_RACK;
+        }
+        return state.getApplicationState(ApplicationState.RACK).value;
+    }
+
+    @Override
+    public final String getDatacenter(InetAddressAndPort endpoint)
+    {
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
+            return getLocalDatacenter();
+        EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
+        if (state == null || state.getApplicationState(ApplicationState.DC) == null)
+        {
+            if (savedEndpoints == null)
+                savedEndpoints = SystemKeyspace.loadDcRackInfo();
+            if (savedEndpoints.containsKey(endpoint))
+                return savedEndpoints.get(endpoint).get("data_center");
+            return DEFAULT_DC;
+        }
+        return state.getApplicationState(ApplicationState.DC).value;
+    }
+}