CASSSIDECAR-216: Capture Metrics for Schema Reporting

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

@@ -24,6 +24,7 @@
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
 
 import com.datastax.driver.core.KeyspaceMetadata;
 import com.datastax.driver.core.TableMetadata;
@@ -32,7 +33,7 @@
 
 /**
  * An abstract class that has to be extended and instantiated for every Cassandra
- * cluster that needs its schema converted into a DataHub-compliant format.
+ * cluster that needs its schema converted into a DataHub-compliant format
  */
 public abstract class IdentifiersProvider
 {
@@ -42,7 +43,26 @@ public abstract class IdentifiersProvider
     protected static final String DATA_PLATFORM_INSTANCE = "dataPlatformInstance";
     protected static final String CONTAINER = "container";
     protected static final String DATASET = "dataset";
-    protected static final String PROD = "PROD";  // DataHub requires this to be {@code PROD} regardless
+    protected static final String PROD = "PROD";  // DataHub requires this to be {@code PROD}
+
+    protected static final ToStringStyle STYLE = new ToStringStyle()
+    {{
+        setUseShortClassName(false);
+        setUseClassName(true);
+        setUseIdentityHashCode(false);
+        setUseFieldNames(false);
+        setContentStart("(");
+        setFieldSeparatorAtStart(false);
+        setFieldSeparator(",");
+        setFieldSeparatorAtEnd(false);
+        setContentEnd(")");
+        setDefaultFullDetail(true);
+        setArrayContentDetail(true);
+        setArrayStart("(");
+        setArraySeparator(",");
+        setArrayEnd(")");
+        setNullText("null");
+    }};
 
     /**
      * A public getter method that returns the name of Cassandra Organization
@@ -231,13 +251,19 @@ public boolean equals(@Nullable Object other)
     @NotNull
     public String toString()
     {
-        return new ToStringBuilder(this)
+        return new ToStringBuilder(this, STYLE)
                 .append(this.organization())
                 .append(this.platform())
                 .append(this.environment())
                 .append(this.application())
                 .append(this.cluster())
                 .append(this.identifier())
-                .toString();
+                .toString()
+                .replaceAll("\\s", "");
+
+        // Use of a custom {@link ToStringStyle} implementation prevents the hash code from being
+        // included into the {@link String} representation; which, in conjunction with the removal
+        // of all whitespace characters, simplifies extraction of {@link IdentifierProvider}
+        // objects from the logs with Splunk; without negatively affecting the readability much
     }
 }

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

@@ -20,9 +20,13 @@
 package org.apache.cassandra.sidecar.datahub;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
+import com.google.common.base.Stopwatch;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Streams;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.KeyspaceMetadata;
@@ -34,6 +38,9 @@
 import datahub.client.Emitter;
 import datahub.event.MetadataChangeProposalWrapper;
 import org.apache.cassandra.sidecar.common.server.utils.ThrowableUtils;
+import org.apache.cassandra.sidecar.metrics.DeltaGauge;
+import org.apache.cassandra.sidecar.metrics.SidecarMetrics;
+import org.apache.cassandra.sidecar.metrics.server.SchemaReportingMetrics;
 import org.jetbrains.annotations.NotNull;
 
 /**
@@ -47,6 +54,8 @@
 @Singleton
 public class SchemaReporter
 {
+    private static final Logger LOGGER = LoggerFactory.getLogger(SchemaReporter.class);
+
     @NotNull
     protected final IdentifiersProvider identifiersProvider;
     @NotNull
@@ -57,6 +66,8 @@ public class SchemaReporter
     protected final List<TableToAspectConverter<? extends RecordTemplate>> tableConverters;
     @NotNull
     protected final EmitterFactory emitterFactory;
+    @NotNull
+    protected final SchemaReportingMetrics reportingMetrics;
 
     /**
      * The public constructor that instantiates {@link SchemaReporter} with default configuration.
@@ -66,10 +77,12 @@ public class SchemaReporter
      *
      * @param identifiersProvider an instance of {@link IdentifiersProvider} to use
      * @param emitterFactory an instance of {@link EmitterFactory} to use
+     * @param sidecarMetrics an instance of {@link SidecarMetrics} to obtain {@link SchemaReportingMetrics} from
      */
     @Inject
     public SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
-                          @NotNull EmitterFactory emitterFactory)
+                          @NotNull EmitterFactory emitterFactory,
+                          @NotNull SidecarMetrics sidecarMetrics)
     {
         this(identifiersProvider,
              ImmutableList.of(new ClusterToDataPlatformInfoConverter(identifiersProvider),
@@ -85,7 +98,8 @@ public SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
                               new TableToDataPlatformInstanceConverter(identifiersProvider),
                               new TableToBrowsePathsV2Converter(identifiersProvider),
                               new TableToBrowsePathsConverter(identifiersProvider)),
-             emitterFactory);
+             emitterFactory,
+             sidecarMetrics.server().schemaReporting());
     }
 
     /**
@@ -96,45 +110,73 @@ public SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
      * @param keyspaceConverters a {@link List} of {@link KeyspaceToAspectConverter} instances to use
      * @param tableConverters a {@link List} of {@link TableToAspectConverter} instances to use
      * @param emitterFactory an instance of {@link EmitterFactory} to use
+     * @param reportingMetrics an instance of {@link SchemaReportingMetrics} to use
      */
     protected SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
                              @NotNull List<ClusterToAspectConverter<? extends RecordTemplate>> clusterConverters,
                              @NotNull List<KeyspaceToAspectConverter<? extends RecordTemplate>> keyspaceConverters,
                              @NotNull List<TableToAspectConverter<? extends RecordTemplate>> tableConverters,
-                             @NotNull EmitterFactory emitterFactory)
+                             @NotNull EmitterFactory emitterFactory,
+                             @NotNull SchemaReportingMetrics reportingMetrics)
     {
         this.identifiersProvider = identifiersProvider;
         this.clusterConverters = clusterConverters;
         this.keyspaceConverters = keyspaceConverters;
         this.tableConverters = tableConverters;
         this.emitterFactory = emitterFactory;
+        this.reportingMetrics = reportingMetrics;
     }
 
     /**
-     * Public method for converting and reporting the Cassandra schema
+     * Public method for converting and reporting the Cassandra schema when triggered by a scheduled periodic task
      *
      * @param cluster the {@link Cluster} to extract Cassandra schema from
      */
-    public void process(@NotNull Cluster cluster)
+    public void processScheduled(@NotNull Cluster cluster)
     {
-        process(cluster.getMetadata());
+        process(cluster.getMetadata(), reportingMetrics.startedSchedule.metric);
     }
 
     /**
-     * Public method for converting and reporting the Cassandra schema
+     * Public method for converting and reporting the Cassandra schema when triggered by a received API request
      *
      * @param metadata the {@link Metadata} to extract Cassandra schema from
      */
-    public void process(@NotNull Metadata metadata)
+    public void processRequested(@NotNull Metadata metadata)
     {
+        process(metadata, reportingMetrics.startedRequest.metric);
+    }
+
+    /**
+     * Private method for converting and reporting the Cassandra schema
+     *
+     * @param metadata the {@link Metadata} to extract Cassandra schema from
+     * @param started the {@link DeltaGauge} for the metric counting invocations
+     */
+    private void process(@NotNull Metadata metadata,
+                         @NotNull DeltaGauge started)
+    {
+        String action = " reporting schema for cluster with identifiers " + identifiersProvider;
+        LOGGER.info("Started" + action);
+        started.increment();
+
         try (Emitter emitter = emitterFactory.emitter())
         {
-            stream(metadata)
-                    .forEach(ThrowableUtils.consumer(emitter::emit));
+            Stopwatch stopwatch = Stopwatch.createStarted();
+            long counter = stream(metadata)
+                    .map(ThrowableUtils.function(emitter::emit))
+                    .count();
+
+            reportingMetrics.durationMilliseconds.metric.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
+            reportingMetrics.sizeAspects.metric.update(counter);
+            reportingMetrics.finishedSuccess.metric.increment();
+            LOGGER.info("Success" + action);
         }
         catch (Exception exception)
         {
-            throw new RuntimeException("Cannot extract schema for cluster " + identifiersProvider.cluster(), exception);
+            reportingMetrics.finishedFailure.metric.increment();
+            LOGGER.error("Failure" + action);
+            throw new RuntimeException(action, exception);
         }
     }
 

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

@@ -89,7 +89,7 @@ public void execute(Promise<Void> promise)
     {
         try
         {
-            reporter.process(session.get().getCluster());
+            reporter.processScheduled(session.get().getCluster());
             promise.complete();
         }
         catch (Throwable throwable)

server/src/main/java/org/apache/cassandra/sidecar/metrics/DeltaGauge.java

@@ -33,7 +33,15 @@ public class DeltaGauge implements Gauge<Long>, Metric
 
     public DeltaGauge()
     {
-        this.count = new AtomicLong();
+        this.count = new AtomicLong(0L);
+    }
+
+    /**
+     * Increments the cumulative value tracked by this {@link DeltaGauge}
+     */
+    public void increment()
+    {
+        count.incrementAndGet();
     }
 
     /**
@@ -55,6 +63,6 @@ public void update(long delta)
     @Override
     public Long getValue()
     {
-        return count.getAndSet(0);
+        return count.getAndSet(0L);
     }
 }

server/src/main/java/org/apache/cassandra/sidecar/metrics/ServerMetrics.java

@@ -19,6 +19,7 @@
 package org.apache.cassandra.sidecar.metrics;
 
 import org.apache.cassandra.sidecar.db.schema.SidecarSchema;
+import org.apache.cassandra.sidecar.metrics.server.SchemaReportingMetrics;
 
 import static org.apache.cassandra.sidecar.metrics.SidecarMetrics.APP_PREFIX;
 
@@ -49,6 +50,11 @@ public interface ServerMetrics
      */
     SchemaMetrics schema();
 
+    /**
+     * @return metrics for the schema reporting
+     */
+    SchemaReportingMetrics schemaReporting();
+
     /**
      * @return metrics related to internal caches that are tracked.
      */

server/src/main/java/org/apache/cassandra/sidecar/metrics/ServerMetricsImpl.java

@@ -21,6 +21,7 @@
 import java.util.Objects;
 
 import com.codahale.metrics.MetricRegistry;
+import org.apache.cassandra.sidecar.metrics.server.SchemaReportingMetrics;
 
 /**
  * {@link ServerMetrics} tracks metrics related to Sidecar server.
@@ -32,6 +33,7 @@ public class ServerMetricsImpl implements ServerMetrics
     protected final ResourceMetrics resourceMetrics;
     protected final RestoreMetrics restoreMetrics;
     protected final SchemaMetrics schemaMetrics;
+    protected final SchemaReportingMetrics schemaReportingMetrics;
     protected final CacheMetrics cacheMetrics;
     protected final CoordinationMetrics coordinationMetrics;
 
@@ -43,6 +45,7 @@ public ServerMetricsImpl(MetricRegistry metricRegistry)
         this.resourceMetrics = new ResourceMetrics(metricRegistry);
         this.restoreMetrics = new RestoreMetrics(metricRegistry);
         this.schemaMetrics = new SchemaMetrics(metricRegistry);
+        this.schemaReportingMetrics = new SchemaReportingMetrics(metricRegistry);
         this.cacheMetrics = new CacheMetrics(metricRegistry);
         this.coordinationMetrics = new CoordinationMetrics(metricRegistry);
     }
@@ -71,6 +74,12 @@ public SchemaMetrics schema()
         return schemaMetrics;
     }
 
+    @Override
+    public SchemaReportingMetrics schemaReporting()
+    {
+        return schemaReportingMetrics;
+    }
+
     @Override
     public CacheMetrics cache()
     {

server/src/main/java/org/apache/cassandra/sidecar/metrics/server/SchemaReportingMetrics.java

@@ -0,0 +1,91 @@
+/*
+ * 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.metrics.server;
+
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.MetricRegistry;
+import org.apache.cassandra.sidecar.metrics.DeltaGauge;
+import org.apache.cassandra.sidecar.metrics.NamedMetric;
+import org.apache.cassandra.sidecar.metrics.ServerMetrics;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks metrics for the schema reporting done by Sidecar
+ */
+public class SchemaReportingMetrics
+{
+    protected static final String DOMAIN = ServerMetrics.SERVER_PREFIX + ".SchemaReporting";
+    protected static final String STARTED = "Started";
+    protected static final String FINISHED = "Finished";
+    protected static final String DURATION = "Duration";
+    protected static final String SIZE = "Size";
+    protected static final String TRIGGER = "Trigger";
+    protected static final String RESULT = "Result";
+    protected static final String UNIT = "Unit";
+    protected static final String REQUEST = "Request";
+    protected static final String SCHEDULE = "Schedule";
+    protected static final String SUCCESS = "Success";
+    protected static final String FAILURE = "Failure";
+    protected static final String ASPECTS = "Aspects";
+    protected static final String MILLISECONDS = "Milliseconds";
+
+    public final NamedMetric<DeltaGauge> startedRequest;
+    public final NamedMetric<DeltaGauge> startedSchedule;
+    public final NamedMetric<DeltaGauge> finishedSuccess;
+    public final NamedMetric<DeltaGauge> finishedFailure;
+    public final NamedMetric<Histogram> durationMilliseconds;
+    public final NamedMetric<Histogram> sizeAspects;
+
+    public SchemaReportingMetrics(@NotNull MetricRegistry registry)
+    {
+        startedSchedule = NamedMetric.builder(name -> registry.gauge(name, DeltaGauge::new))
+                .withDomain(DOMAIN)
+                .withName(STARTED)
+                .addTag(TRIGGER, SCHEDULE)
+                .build();
+        startedRequest = NamedMetric.builder(name -> registry.gauge(name, DeltaGauge::new))
+                .withDomain(DOMAIN)
+                .withName(STARTED)
+                .addTag(TRIGGER, REQUEST)
+                .build();
+
+        finishedSuccess = NamedMetric.builder(name -> registry.gauge(name, DeltaGauge::new))
+                .withDomain(DOMAIN)
+                .withName(FINISHED)
+                .addTag(RESULT, SUCCESS)
+                .build();
+        finishedFailure = NamedMetric.builder(name -> registry.gauge(name, DeltaGauge::new))
+                .withDomain(DOMAIN)
+                .withName(FINISHED)
+                .addTag(RESULT, FAILURE)
+                .build();
+
+        sizeAspects = NamedMetric.builder(registry::histogram)
+                .withDomain(DOMAIN)
+                .withName(SIZE)
+                .addTag(UNIT, ASPECTS)
+                .build();
+
+        durationMilliseconds = NamedMetric.builder(registry::histogram)
+                .withDomain(DOMAIN)
+                .withName(DURATION)
+                .addTag(UNIT, MILLISECONDS)
+                .build();
+    }
+}