CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report

CHANGES.txt

@@ -1,5 +1,6 @@
 1.0.0
 -----
+ * Create Endpoint that Triggers an Immediate Schema Report (CASSSIDECAR-203)
  * Adapt to cluster topology change for restore jobs (CASSSIDECAR-185)
  * Fix PeriodicTaskExecutor double execution due to race from reschedule (CASSSIDECAR-210)
  * Upgrade Netty to 4.1.118.Final and Vert.x to 4.5.13 Version (CASSSIDECAR-207)
@@ -8,7 +9,7 @@
  * Sidecar schema initialization can be executed on multiple thread (CASSSIDECAR-200)
  * Make sidecar operations resilient to down Cassandra nodes (CASSSIDECAR-201)
  * Fix Cassandra instance not found error (CASSSIDECAR-192)
- * Implemented Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
+ * Implement Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
  * Add sidecar endpoint to retrieve stream stats (CASSSIDECAR-180)
  * Add sidecar endpoint to retrieve cassandra gossip health (CASSSIDECAR-173)
  * Fix SidecarSchema stuck at initialization due to ClusterLeaseTask scheduling (CASSSIDECAR-189)

client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java

@@ -121,6 +121,9 @@ public final class ApiEndpointsV1
     public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
     public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
 
+    // Schema Reporting
+    private static final String REPORT_SCHEMA = "/report-schema";
+    public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
 
     public static final String CONNECTED_CLIENT_STATS_ROUTE = API_V1 + CASSANDRA + "/stats/connected-clients";
 

server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java

@@ -63,6 +63,9 @@ public class BasicPermissions
     public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
     public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
 
+    // Permissions related to Schema Reporting
+    public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
+
     // cassandra cluster related permissions
     public static final Permission READ_SCHEMA = new DomainAwarePermission("SCHEMA:READ", CLUSTER_SCOPE);
     public static final Permission READ_SCHEMA_KEYSPACE_SCOPED = new DomainAwarePermission("SCHEMA:READ", KEYSPACE_SCOPE);

server/src/main/java/org/apache/cassandra/sidecar/datahub/SchemaReporter.java

@@ -113,13 +113,23 @@ protected SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
     /**
      * Public method for converting and reporting the Cassandra schema
      *
-     * @param cluster a {@link Cluster} to extract Cassandra schema from
+     * @param cluster the {@link Cluster} to extract Cassandra schema from
      */
     public void process(@NotNull Cluster cluster)
+    {
+        process(cluster.getMetadata());
+    }
+
+    /**
+     * Public method for converting and reporting the Cassandra schema
+     *
+     * @param metadata the {@link Metadata} to extract Cassandra schema from
+     */
+    public void process(@NotNull Metadata metadata)
     {
         try (Emitter emitter = emitterFactory.emitter())
         {
-            stream(cluster.getMetadata())
+            stream(metadata)
                     .forEach(ThrowableUtils.consumer(emitter::emit));
         }
         catch (Exception exception)

server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java

@@ -0,0 +1,101 @@
+/*
+ * 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.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+    @NotNull
+    private final SchemaReporter schemaReporter;
+
+    /**
+     * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+     * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+     *
+     * @param metadata the metadata fetcher
+     * @param executor executor pools for blocking executions
+     * @param reporter executor pools for blocking executions
+     */
+    @Inject
+    public ReportSchemaHandler(@NotNull InstanceMetadataFetcher metadata,
+                               @NotNull ExecutorPools executor,
+                               @NotNull SchemaReporter reporter)
+    {
+        super(metadata, executor, null);
+
+        schemaReporter = reporter;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @NotNull
+    public Set<Authorization> requiredAuthorizations()
+    {
+        return Collections.singleton(BasicPermissions.REPORT_SCHEMA.toAuthorization());
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @Nullable
+    protected Void extractParamsOrThrow(@NotNull RoutingContext context)
+    {
+        return null;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    protected void handleInternal(@NotNull RoutingContext context,
+                                  @NotNull HttpServerRequest http,
+                                  @NotNull String host,
+                                  @NotNull SocketAddress address,
+                                  @Nullable Void request)
+    {
+        executorPools.service()
+                     .runBlocking(() -> metadataFetcher.runOnFirstAvailableInstance(instance ->
+                            schemaReporter.process(instance.delegate().metadata())))
+                     .onSuccess(context::json)
+                     .onFailure(throwable -> processFailure(throwable, context, host, address, request));
+    }
+}

server/src/main/java/org/apache/cassandra/sidecar/server/MainModule.java

@@ -131,6 +131,7 @@
 import org.apache.cassandra.sidecar.routes.ListOperationalJobsHandler;
 import org.apache.cassandra.sidecar.routes.NodeDecommissionHandler;
 import org.apache.cassandra.sidecar.routes.OperationalJobHandler;
+import org.apache.cassandra.sidecar.routes.ReportSchemaHandler;
 import org.apache.cassandra.sidecar.routes.RingHandler;
 import org.apache.cassandra.sidecar.routes.RoutingOrder;
 import org.apache.cassandra.sidecar.routes.SchemaHandler;
@@ -341,6 +342,7 @@ public Router vertxRouter(Vertx vertx,
                               SSTableCleanupHandler ssTableCleanupHandler,
                               StreamCdcSegmentHandler streamCdcSegmentHandler,
                               ListCdcDirHandler listCdcDirHandler,
+                              ReportSchemaHandler reportSchemaHandler,
                               RestoreRequestValidationHandler validateRestoreJobRequest,
                               DiskSpaceProtectionHandler diskSpaceProtection,
                               ValidateTableExistenceHandler validateTableExistence,
@@ -598,6 +600,14 @@ public Router vertxRouter(Vertx vertx,
                                     .handler(streamCdcSegmentHandler)
                                     .build();
 
+        // Schema Reporting
+        protectedRouteBuilderFactory.get()
+                                    .router(router)
+                                    .method(HttpMethod.GET)
+                                    .endpoint(ApiEndpointsV1.REPORT_SCHEMA_ROUTE)
+                                    .handler(reportSchemaHandler)
+                                    .build();
+
         return router;
     }
 
@@ -902,7 +912,8 @@ public IdentifiersProvider identifiersProvider(@NotNull InstanceMetadataFetcher
             @NotNull
             protected String initialize()
             {
-                return fetcher.callOnFirstAvailableInstance(i -> i.delegate().storageOperations().clusterName());
+                return fetcher.callOnFirstAvailableInstance(instance ->
+                        instance.delegate().storageOperations().clusterName());
             }
         };
 

server/src/main/java/org/apache/cassandra/sidecar/utils/InstanceMetadataFetcher.java

@@ -19,8 +19,8 @@
 package org.apache.cassandra.sidecar.utils;
 
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.function.Function;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -97,13 +97,29 @@ public CassandraAdapterDelegate delegate(@NotNull String host) throws NoSuchCass
     }
 
     /**
-     * Iterate through the local instances and call the function on the first available instance, i.e. no CassandraUnavailableException
-     * or OperationUnavailableException is thrown for the operations
+     * Iterate through the local instances and run the {@link Consumer} on the first available one,
+     * so no {@link CassandraUnavailableException} or {@link OperationUnavailableException} is thrown for the operations
+     *
+     * @param consumer a {@link Consumer} that processes {@link InstanceMetadata} and returns no result
+     * @throws CassandraUnavailableException if all local instances were exhausted
+     */
+    public void runOnFirstAvailableInstance(Consumer<InstanceMetadata> consumer) throws CassandraUnavailableException
+    {
+        callOnFirstAvailableInstance(metadata ->
+        {
+            consumer.accept(metadata);
+            return null;
+        });
+    }
+
+    /**
+     * Iterate through the local instances and call the {@link Function} on the first available one,
+     * so no {@link CassandraUnavailableException} or {@link OperationUnavailableException} is thrown for the operations
      *
-     * @param function function applies to {@link InstanceMetadata}
-     * @return function eval result. Null can be returned when all local instances are exhausted
      * @param <T> type of the result
-     * @throws CassandraUnavailableException when all local instances are exhausted.
+     * @param function a {@link Function} that maps {@link InstanceMetadata} to {@link T}
+     * @return evaluation result of the {@code function}; can be {@code null} if all local instances were exhausted
+     * @throws CassandraUnavailableException if all local instances were exhausted
      */
     @NotNull
     public <T> T callOnFirstAvailableInstance(Function<InstanceMetadata, T> function) throws CassandraUnavailableException