Add diagnostic events base classes

patch by Stefan Podkowinski; reviewed by Mick Semb Wever for CASSANDRA-13457

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0
+ * Add base classes for diagnostic events (CASSANDRA-13457)
  * Clear view system metadata when dropping keyspace (CASSANDRA-14646)
  * Allocate ReentrantLock on-demand in java11 AtomicBTreePartitionerBase (CASSANDRA-14637)
  * Make all existing virtual tables use LocalPartitioner (CASSANDRA-14640)

conf/cassandra.yaml

@@ -1217,3 +1217,8 @@ audit_logging_options:
 # validate tombstones on reads and compaction
 # can be either "disabled", "warn" or "exception"
 # corrupted_tombstone_strategy: disabled
+
+# Diagnostic Events #
+# If enabled, diagnostic events can be helpful for troubleshooting operational issues. Emitted events contain details
+# on internal state and temporal relationships across events, accessible by clients via JMX.
+diagnostic_events_enabled: false

src/java/org/apache/cassandra/config/Config.java

@@ -386,6 +386,9 @@ public class Config
     public volatile AuditLogOptions audit_logging_options = new AuditLogOptions();
 
     public CorruptedTombstoneStrategy corrupted_tombstone_strategy = CorruptedTombstoneStrategy.disabled;
+
+    public volatile boolean diagnostic_events_enabled = false;
+
     /**
      * @deprecated migrate to {@link DatabaseDescriptor#isClientInitialized()}
      */

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -2536,6 +2536,17 @@ public class DatabaseDescriptor
         return conf.back_pressure_enabled;
     }
 
+    public static boolean diagnosticEventsEnabled()
+    {
+        return conf.diagnostic_events_enabled;
+    }
+
+    @VisibleForTesting
+    public static void setDiagnosticEventsEnabled(boolean enabled)
+    {
+        conf.diagnostic_events_enabled = enabled;
+    }
+
     @VisibleForTesting
     public static void setBackPressureStrategy(BackPressureStrategy strategy)
     {

src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java

@@ -95,6 +95,11 @@ public final class AlterKeyspaceStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.ALTER_KEYSPACE, keyspaceName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s)", getClass().getSimpleName(), keyspaceName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final String keyspaceName;

src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java

@@ -81,6 +81,11 @@ public abstract class AlterTableStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.ALTER_TABLE, keyspaceName, tableName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, tableName);
+    }
+
     abstract KeyspaceMetadata apply(KeyspaceMetadata keyspace, TableMetadata table);
 
     /**

src/java/org/apache/cassandra/cql3/statements/schema/AlterTypeStatement.java

@@ -83,6 +83,11 @@ public abstract class AlterTypeStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.ALTER_TYPE, keyspaceName, typeName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, typeName);
+    }
+
     private static final class AddField extends AlterTypeStatement
     {
         private final FieldIdentifier fieldName;

src/java/org/apache/cassandra/cql3/statements/schema/AlterViewStatement.java

@@ -92,6 +92,11 @@ public final class AlterViewStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.ALTER_VIEW, keyspaceName, viewName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, viewName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final QualifiedName name;

src/java/org/apache/cassandra/cql3/statements/schema/CreateAggregateStatement.java

@@ -271,6 +271,11 @@ public final class CreateAggregateStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.CREATE_AGGREGATE, keyspaceName, aggregateName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, aggregateName);
+    }
+
     private String stateFunctionString()
     {
         return format("%s(%s)", stateFunctionName, join(", ", transform(concat(singleton(rawStateType), rawArgumentTypes), Object::toString)));

src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java

@@ -198,6 +198,11 @@ public final class CreateFunctionStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.CREATE_FUNCTION, keyspaceName, functionName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, functionName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final FunctionName name;

src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java

@@ -194,6 +194,11 @@ public final class CreateIndexStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.CREATE_INDEX, keyspaceName, indexName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, indexName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final QualifiedName tableName;

src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java

@@ -94,6 +94,11 @@ public final class CreateKeyspaceStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.CREATE_KEYSPACE, keyspaceName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s)", getClass().getSimpleName(), keyspaceName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         public final String keyspaceName;

src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java

@@ -123,6 +123,11 @@ public final class CreateTableStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.CREATE_TABLE, keyspaceName, tableName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, tableName);
+    }
+
     public TableMetadata.Builder builder(Types types)
     {
         attrs.validate();

src/java/org/apache/cassandra/cql3/statements/schema/CreateTriggerStatement.java

@@ -96,6 +96,11 @@ public final class CreateTriggerStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.CREATE_TRIGGER, keyspaceName, triggerName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, triggerName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final QualifiedName tableName;

src/java/org/apache/cassandra/cql3/statements/schema/CreateTypeStatement.java

@@ -115,6 +115,11 @@ public final class CreateTypeStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.CREATE_TYPE, keyspaceName, typeName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, typeName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final UTName name;

src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java

@@ -341,6 +341,11 @@ public final class CreateViewStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.CREATE_VIEW, keyspaceName, viewName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, viewName);
+    }
+
     public final static class Raw extends CQLStatement.Raw
     {
         private final QualifiedName tableName;

src/java/org/apache/cassandra/cql3/statements/schema/DropAggregateStatement.java

@@ -139,6 +139,11 @@ public final class DropAggregateStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.DROP_AGGREGATE, keyspaceName, aggregateName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, aggregateName);
+    }
+
     private List<AbstractType<?>> prepareArgumentTypes(Types types)
     {
         return arguments.stream()

src/java/org/apache/cassandra/cql3/statements/schema/DropFunctionStatement.java

@@ -147,6 +147,11 @@ public final class DropFunctionStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.DROP_FUNCTION, keyspaceName, functionName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, functionName);
+    }
+
     private List<AbstractType<?>> prepareArgumentTypes(Types types)
     {
         return arguments.stream()

src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java

@@ -90,6 +90,11 @@ public final class DropIndexStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.DROP_INDEX, keyspaceName, indexName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, indexName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final QualifiedName name;

src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java

@@ -64,6 +64,11 @@ public final class DropKeyspaceStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.DROP_KEYSPACE, keyspaceName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s)", getClass().getSimpleName(), keyspaceName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final String keyspaceName;

src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java

@@ -92,6 +92,11 @@ public final class DropTableStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.DROP_TABLE, keyspaceName, tableName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, tableName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final QualifiedName name;

src/java/org/apache/cassandra/cql3/statements/schema/DropTriggerStatement.java

@@ -82,6 +82,11 @@ public final class DropTriggerStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.DROP_TRIGGER, keyspaceName, triggerName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, triggerName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final QualifiedName tableName;

src/java/org/apache/cassandra/cql3/statements/schema/DropTypeStatement.java

@@ -129,6 +129,11 @@ public final class DropTypeStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.DROP_TYPE, keyspaceName, typeName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, typeName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final UTName name;

src/java/org/apache/cassandra/cql3/statements/schema/DropViewStatement.java

@@ -78,6 +78,11 @@ public final class DropViewStatement extends AlterSchemaStatement
         return new AuditLogContext(AuditLogEntryType.DROP_VIEW, keyspaceName, viewName);
     }
 
+    public String toString()
+    {
+        return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, viewName);
+    }
+
     public static final class Raw extends CQLStatement.Raw
     {
         private final QualifiedName name;

src/java/org/apache/cassandra/dht/BootStrapper.java

@@ -168,7 +168,11 @@ public class BootStrapper extends ProgressEventNotifierSupport
 
         // if user specified tokens, use those
         if (initialTokens.size() > 0)
-            return getSpecifiedTokens(metadata, initialTokens);
+        {
+            Collection<Token> tokens = getSpecifiedTokens(metadata, initialTokens);
+            BootstrapDiagnostics.useSpecifiedTokens(address, allocationKeyspace, tokens, DatabaseDescriptor.getNumTokens());
+            return tokens;
+        }
 
         int numTokens = DatabaseDescriptor.getNumTokens();
         if (numTokens < 1)
@@ -180,7 +184,9 @@ public class BootStrapper extends ProgressEventNotifierSupport
         if (numTokens == 1)
             logger.warn("Picking random token for a single vnode.  You should probably add more vnodes and/or use the automatic token allocation mechanism.");
 
-        return getRandomTokens(metadata, numTokens);
+        Collection<Token> tokens = getRandomTokens(metadata, numTokens);
+        BootstrapDiagnostics.useRandomTokens(address, metadata, numTokens, tokens);
+        return tokens;
     }
 
     private static Collection<Token> getSpecifiedTokens(final TokenMetadata metadata,
@@ -213,7 +219,9 @@ public class BootStrapper extends ProgressEventNotifierSupport
             throw new ConfigurationException("Problem opening token allocation keyspace " + allocationKeyspace);
         AbstractReplicationStrategy rs = ks.getReplicationStrategy();
 
-        return TokenAllocation.allocateTokens(metadata, rs, address, numTokens);
+        Collection<Token> tokens = TokenAllocation.allocateTokens(metadata, rs, address, numTokens);
+        BootstrapDiagnostics.tokensAllocated(address, metadata, allocationKeyspace, numTokens, tokens);
+        return tokens;
     }
 
     public static Collection<Token> getRandomTokens(TokenMetadata metadata, int numTokens)

src/java/org/apache/cassandra/dht/BootstrapDiagnostics.java

@@ -0,0 +1,80 @@
+/*
+ * 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.dht;
+
+import java.util.Collection;
+import com.google.common.collect.ImmutableList;
+
+import org.apache.cassandra.dht.BootstrapEvent.BootstrapEventType;
+import org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.TokenMetadata;
+
+/**
+ * Utility methods for bootstrap related activities.
+ */
+final class BootstrapDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private BootstrapDiagnostics()
+    {
+    }
+
+    static void useSpecifiedTokens(InetAddressAndPort address, String allocationKeyspace, Collection<Token> initialTokens,
+                                   int numTokens)
+    {
+        if (isEnabled(BootstrapEventType.BOOTSTRAP_USING_SPECIFIED_TOKENS))
+            service.publish(new BootstrapEvent(BootstrapEventType.BOOTSTRAP_USING_SPECIFIED_TOKENS,
+                                               address,
+                                               null,
+                                               allocationKeyspace,
+                                               numTokens,
+                                               ImmutableList.copyOf(initialTokens)));
+    }
+
+    static void useRandomTokens(InetAddressAndPort address, TokenMetadata metadata, int numTokens, Collection<Token> tokens)
+    {
+        if (isEnabled(BootstrapEventType.BOOTSTRAP_USING_RANDOM_TOKENS))
+            service.publish(new BootstrapEvent(BootstrapEventType.BOOTSTRAP_USING_RANDOM_TOKENS,
+                                               address,
+                                               metadata.cloneOnlyTokenMap(),
+                                               null,
+                                               numTokens,
+                                               ImmutableList.copyOf(tokens)));
+    }
+
+    static void tokensAllocated(InetAddressAndPort address, TokenMetadata metadata,
+                                String allocationKeyspace, int numTokens, Collection<Token> tokens)
+    {
+        if (isEnabled(BootstrapEventType.TOKENS_ALLOCATED))
+            service.publish(new BootstrapEvent(BootstrapEventType.TOKENS_ALLOCATED,
+                                               address,
+                                               metadata.cloneOnlyTokenMap(),
+                                               allocationKeyspace,
+                                               numTokens,
+                                               ImmutableList.copyOf(tokens)));
+    }
+
+    private static boolean isEnabled(BootstrapEventType type)
+    {
+        return service.isEnabled(BootstrapEvent.class, type);
+    }
+
+}

src/java/org/apache/cassandra/dht/BootstrapEvent.java

@@ -0,0 +1,82 @@
+/*
+ * 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.dht;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableCollection;
+
+import org.apache.cassandra.diag.DiagnosticEvent;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.TokenMetadata;
+
+/**
+ * DiagnosticEvent implementation for bootstrap related activities.
+ */
+final class BootstrapEvent extends DiagnosticEvent
+{
+
+    private final BootstrapEventType type;
+    @Nullable
+    private final TokenMetadata tokenMetadata;
+    private final InetAddressAndPort address;
+    @Nullable
+    private final String allocationKeyspace;
+    private final Integer numTokens;
+    private final Collection<Token> tokens;
+
+    BootstrapEvent(BootstrapEventType type, InetAddressAndPort address, @Nullable TokenMetadata tokenMetadata,
+                   @Nullable String allocationKeyspace, int numTokens, ImmutableCollection<Token> tokens)
+    {
+        this.type = type;
+        this.address = address;
+        this.tokenMetadata = tokenMetadata;
+        this.allocationKeyspace = allocationKeyspace;
+        this.numTokens = numTokens;
+        this.tokens = tokens;
+    }
+
+    enum BootstrapEventType
+    {
+        BOOTSTRAP_USING_SPECIFIED_TOKENS,
+        BOOTSTRAP_USING_RANDOM_TOKENS,
+        TOKENS_ALLOCATED
+    }
+
+
+    public BootstrapEventType getType()
+    {
+        return type;
+    }
+
+    public Map<String, Serializable> toMap()
+    {
+        // be extra defensive against nulls and bugs
+        HashMap<String, Serializable> ret = new HashMap<>();
+        ret.put("tokenMetadata", String.valueOf(tokenMetadata));
+        ret.put("allocationKeyspace", allocationKeyspace);
+        ret.put("numTokens", numTokens);
+        ret.put("tokens", String.valueOf(tokens));
+        return ret;
+    }
+}

src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java

@@ -86,6 +86,7 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
             sortedUnits.add(new Weighted<UnitInfo>(unitInfo.ownership, unitInfo));
         }
 
+        TokenAllocatorDiagnostics.tokenInfosCreated(this, sortedUnits, sortedTokens, first);
         return first;
     }
 
@@ -127,6 +128,7 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
         }
         unitInfos.put(newUnit.unit, newUnit);
         createTokenInfos(unitInfos);
+        TokenAllocatorDiagnostics.randomTokensGenerated(this, numTokens, sortedUnits, sortedTokens, newUnit.unit, tokens);
         return tokens;
     }
 
@@ -232,6 +234,7 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
         }
         sortedUnits.add(new Weighted<>(newUnitInfo.ownership, newUnitInfo));
 
+        TokenAllocatorDiagnostics.unitedAdded(this, numTokens, sortedUnits, sortedTokens, newTokens, newUnit);
         return newTokens;
     }
 
@@ -257,6 +260,7 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
             tokens.add(tokenInfo.value.token);
         }
         sortedTokens.keySet().removeAll(tokens);
+        TokenAllocatorDiagnostics.unitRemoved(this, n, sortedUnits, sortedTokens);
     }
 
     public int getReplicas()

src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java

@@ -132,7 +132,9 @@ class ReplicationAwareTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
             }
         }
 
-        return ImmutableList.copyOf(unitToTokens.get(newUnit));
+        ImmutableList<Token> newTokens = ImmutableList.copyOf(unitToTokens.get(newUnit));
+        TokenAllocatorDiagnostics.unitedAdded(this, numTokens, unitToTokens, sortedTokens, newTokens, newUnit);
+        return newTokens;
     }
 
     private Collection<Token> generateRandomTokens(Unit newUnit, int numTokens)
@@ -148,6 +150,7 @@ class ReplicationAwareTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
                 unitToTokens.put(newUnit, token);
             }
         }
+        TokenAllocatorDiagnostics.randomTokensGenerated(this, numTokens, unitToTokens, sortedTokens, newUnit, tokens);
         return tokens;
     }
 
@@ -176,6 +179,7 @@ class ReplicationAwareTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
             curr = curr.next;
         } while (curr != first);
 
+        TokenAllocatorDiagnostics.tokenInfosCreated(this, unitToTokens, first);
         return first;
     }
 
@@ -526,6 +530,7 @@ class ReplicationAwareTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
     {
         Collection<Token> tokens = unitToTokens.removeAll(n);
         sortedTokens.keySet().removeAll(tokens);
+        TokenAllocatorDiagnostics.unitRemoved(this, n, unitToTokens, sortedTokens);
     }
 
     public int unitCount()

src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorDiagnostics.java

@@ -0,0 +1,195 @@
+/*
+ * 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.dht.tokenallocator;
+
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Queue;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Multimap;
+
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.dht.tokenallocator.TokenAllocatorBase.TokenInfo;
+import org.apache.cassandra.dht.tokenallocator.TokenAllocatorBase.UnitInfo;
+import org.apache.cassandra.dht.tokenallocator.TokenAllocatorBase.Weighted;
+import org.apache.cassandra.dht.tokenallocator.TokenAllocatorEvent.TokenAllocatorEventType;
+import org.apache.cassandra.diag.DiagnosticEventService;
+
+/**
+ * Utility methods for DiagnosticEvent around {@link TokenAllocator} activities.
+ */
+final class TokenAllocatorDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private TokenAllocatorDiagnostics()
+    {
+    }
+
+    static <Unit> void noReplicationTokenAllocatorInstanciated(NoReplicationTokenAllocator<Unit> allocator)
+    {
+        if (isEnabled(TokenAllocatorEventType.NO_REPLICATION_AWARE_TOKEN_ALLOCATOR_INSTANCIATED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.NO_REPLICATION_AWARE_TOKEN_ALLOCATOR_INSTANCIATED,
+                                                      allocator, null, null, null, null, null, null, null));
+    }
+
+    static <Unit> void replicationTokenAllocatorInstanciated(ReplicationAwareTokenAllocator<Unit> allocator)
+    {
+        if (isEnabled(TokenAllocatorEventType.REPLICATION_AWARE_TOKEN_ALLOCATOR_INSTANCIATED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.REPLICATION_AWARE_TOKEN_ALLOCATOR_INSTANCIATED,
+                                                      allocator, null, null, null,null, null, null, null));
+    }
+
+    static <Unit> void unitedAdded(TokenAllocatorBase<Unit> allocator, int numTokens,
+                                   Queue<Weighted<UnitInfo>> sortedUnits, NavigableMap<Token, Unit> sortedTokens,
+                                   List<Token> tokens, Unit unit)
+    {
+        if (isEnabled(TokenAllocatorEventType.UNIT_ADDED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.UNIT_ADDED,
+                                                      allocator,
+                                                      numTokens,
+                                                      ImmutableList.copyOf(sortedUnits),
+                                                      null,
+                                                      ImmutableMap.copyOf(sortedTokens),
+                                                      ImmutableList.copyOf(tokens),
+                                                      unit,
+                                                      null));
+    }
+
+    static <Unit> void unitedAdded(TokenAllocatorBase<Unit> allocator, int numTokens,
+                                   Multimap<Unit, Token> unitToTokens, NavigableMap<Token, Unit> sortedTokens,
+                                   List<Token> tokens, Unit unit)
+    {
+        if (isEnabled(TokenAllocatorEventType.UNIT_ADDED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.UNIT_ADDED,
+                                                      allocator,
+                                                      numTokens,
+                                                      null,
+                                                      ImmutableMap.copyOf(unitToTokens.asMap()),
+                                                      ImmutableMap.copyOf(sortedTokens),
+                                                      ImmutableList.copyOf(tokens),
+                                                      unit,
+                                                      null));
+    }
+
+
+    static <Unit> void unitRemoved(TokenAllocatorBase<Unit> allocator, Unit unit,
+                                   Queue<Weighted<UnitInfo>> sortedUnits, Map<Token, Unit> sortedTokens)
+    {
+        if (isEnabled(TokenAllocatorEventType.UNIT_REMOVED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.UNIT_REMOVED,
+                                                      allocator,
+                                                      null,
+                                                      ImmutableList.copyOf(sortedUnits),
+                                                      null,
+                                                      ImmutableMap.copyOf(sortedTokens),
+                                                      null,
+                                                      unit,
+                                                      null));
+    }
+
+    static <Unit> void unitRemoved(TokenAllocatorBase<Unit> allocator, Unit unit,
+                                   Multimap<Unit, Token> unitToTokens, Map<Token, Unit> sortedTokens)
+    {
+        if (isEnabled(TokenAllocatorEventType.UNIT_REMOVED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.UNIT_REMOVED,
+                                                      allocator,
+                                                      null,
+                                                      null,
+                                                      ImmutableMap.copyOf(unitToTokens.asMap()),
+                                                      ImmutableMap.copyOf(sortedTokens),
+                                                      null,
+                                                      unit,
+                                                      null));
+    }
+
+    static <Unit> void tokenInfosCreated(TokenAllocatorBase<Unit> allocator, Queue<Weighted<UnitInfo>> sortedUnits,
+                                         Map<Token, Unit> sortedTokens, TokenInfo<Unit> tokenInfo)
+    {
+        if (isEnabled(TokenAllocatorEventType.TOKEN_INFOS_CREATED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.TOKEN_INFOS_CREATED,
+                                                      allocator,
+                                                      null,
+                                                      ImmutableList.copyOf(sortedUnits),
+                                                      null,
+                                                      ImmutableMap.copyOf(sortedTokens),
+                                                      null,
+                                                      null,
+                                                      tokenInfo));
+    }
+
+    static <Unit> void tokenInfosCreated(TokenAllocatorBase<Unit> allocator, Multimap<Unit, Token> unitToTokens,
+                                         TokenInfo<Unit> tokenInfo)
+    {
+        if (isEnabled(TokenAllocatorEventType.TOKEN_INFOS_CREATED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.TOKEN_INFOS_CREATED,
+                                                      allocator,
+                                                      null,
+                                                      null,
+                                                      ImmutableMap.copyOf(unitToTokens.asMap()),
+                                                      null,
+                                                      null,
+                                                      null,
+                                                      tokenInfo));
+    }
+
+    static <Unit> void randomTokensGenerated(TokenAllocatorBase<Unit> allocator,
+                                             int numTokens, Queue<Weighted<UnitInfo>> sortedUnits,
+                                             NavigableMap<Token, Unit> sortedTokens, Unit newUnit,
+                                             Set<Token> tokens)
+    {
+        if (isEnabled(TokenAllocatorEventType.RANDOM_TOKENS_GENERATED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.RANDOM_TOKENS_GENERATED,
+                                                      allocator,
+                                                      numTokens,
+                                                      ImmutableList.copyOf(sortedUnits),
+                                                      null,
+                                                      ImmutableMap.copyOf(sortedTokens),
+                                                      ImmutableList.copyOf(tokens),
+                                                      newUnit,
+                                                      null));
+    }
+
+    static <Unit> void randomTokensGenerated(TokenAllocatorBase<Unit> allocator,
+                                             int numTokens, Multimap<Unit, Token> unitToTokens,
+                                             NavigableMap<Token, Unit> sortedTokens, Unit newUnit,
+                                             Set<Token> tokens)
+    {
+        if (isEnabled(TokenAllocatorEventType.RANDOM_TOKENS_GENERATED))
+            service.publish(new TokenAllocatorEvent<>(TokenAllocatorEventType.RANDOM_TOKENS_GENERATED,
+                                                      allocator,
+                                                      numTokens,
+                                                      null,
+                                                      ImmutableMap.copyOf(unitToTokens.asMap()),
+                                                      ImmutableMap.copyOf(sortedTokens),
+                                                      ImmutableList.copyOf(tokens),
+                                                      newUnit,
+                                                      null));
+    }
+
+    private static boolean isEnabled(TokenAllocatorEventType type)
+    {
+        return service.isEnabled(TokenAllocatorEvent.class, type);
+    }
+
+}

src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorEvent.java

@@ -0,0 +1,113 @@
+/*
+ * 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.dht.tokenallocator;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.dht.tokenallocator.TokenAllocatorBase.TokenInfo;
+import org.apache.cassandra.dht.tokenallocator.TokenAllocatorBase.UnitInfo;
+import org.apache.cassandra.dht.tokenallocator.TokenAllocatorBase.Weighted;
+import org.apache.cassandra.diag.DiagnosticEvent;
+
+/**
+ * DiagnosticEvent implementation for {@link TokenAllocator} activities.
+ */
+final class TokenAllocatorEvent<Unit> extends DiagnosticEvent
+{
+
+    private final TokenAllocatorEventType type;
+    private final TokenAllocatorBase<Unit> allocator;
+    private final int replicas;
+    @Nullable
+    private final Integer numTokens;
+    @Nullable
+    private final Collection<Weighted<UnitInfo>> sortedUnits;
+    @Nullable
+    private final Map<Unit, Collection<Token>> unitToTokens;
+    @Nullable
+    private final ImmutableMap<Token, Unit> sortedTokens;
+    @Nullable
+    private final List<Token> tokens;
+    @Nullable
+    private final Unit unit;
+    @Nullable
+    private final TokenInfo<Unit> tokenInfo;
+
+    TokenAllocatorEvent(TokenAllocatorEventType type, TokenAllocatorBase<Unit> allocator, @Nullable Integer numTokens,
+                        @Nullable ImmutableList<Weighted<UnitInfo>> sortedUnits, @Nullable ImmutableMap<Unit, Collection<Token>> unitToTokens,
+                        @Nullable ImmutableMap<Token, Unit> sortedTokens, @Nullable ImmutableList<Token> tokens, Unit unit,
+                        @Nullable TokenInfo<Unit> tokenInfo)
+    {
+        this.type = type;
+        this.allocator = allocator;
+        this.replicas = allocator.getReplicas();
+        this.numTokens = numTokens;
+        this.sortedUnits = sortedUnits;
+        this.unitToTokens = unitToTokens;
+        this.sortedTokens = sortedTokens;
+        this.tokens = tokens;
+        this.unit = unit;
+        this.tokenInfo = tokenInfo;
+    }
+
+    enum TokenAllocatorEventType
+    {
+        REPLICATION_AWARE_TOKEN_ALLOCATOR_INSTANCIATED,
+        NO_REPLICATION_AWARE_TOKEN_ALLOCATOR_INSTANCIATED,
+        UNIT_ADDED,
+        UNIT_REMOVED,
+        TOKEN_INFOS_CREATED,
+        RANDOM_TOKENS_GENERATED,
+        TOKENS_ALLOCATED
+    }
+
+    public TokenAllocatorEventType getType()
+    {
+        return type;
+    }
+
+    public HashMap<String, Serializable> toMap()
+    {
+        // be extra defensive against nulls and bugs
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (allocator != null)
+        {
+            if (allocator.partitioner != null) ret.put("partitioner", allocator.partitioner.getClass().getSimpleName());
+            if (allocator.strategy != null) ret.put("strategy", allocator.strategy.getClass().getSimpleName());
+        }
+        ret.put("replicas", replicas);
+        ret.put("numTokens", this.numTokens);
+        ret.put("sortedUnits", String.valueOf(sortedUnits));
+        ret.put("sortedTokens", String.valueOf(sortedTokens));
+        ret.put("unitToTokens", String.valueOf(unitToTokens));
+        ret.put("tokens", String.valueOf(tokens));
+        ret.put("unit", String.valueOf(unit));
+        ret.put("tokenInfo", String.valueOf(tokenInfo));
+        return ret;
+    }
+}

src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorFactory.java

@@ -37,9 +37,13 @@ public class TokenAllocatorFactory
         if(strategy.replicas() == 1)
         {
             logger.info("Using NoReplicationTokenAllocator.");
-            return new NoReplicationTokenAllocator<>(sortedTokens, strategy, partitioner);
+            NoReplicationTokenAllocator<InetAddressAndPort> allocator = new NoReplicationTokenAllocator<>(sortedTokens, strategy, partitioner);
+            TokenAllocatorDiagnostics.noReplicationTokenAllocatorInstanciated(allocator);
+            return allocator;
         }
         logger.info("Using ReplicationAwareTokenAllocator.");
-        return new ReplicationAwareTokenAllocator<>(sortedTokens, strategy, partitioner);
+        ReplicationAwareTokenAllocator<InetAddressAndPort> allocator = new ReplicationAwareTokenAllocator<>(sortedTokens, strategy, partitioner);
+        TokenAllocatorDiagnostics.replicationTokenAllocatorInstanciated(allocator);
+        return allocator;
     }
 }

src/java/org/apache/cassandra/diag/DiagnosticEvent.java

@@ -0,0 +1,50 @@
+/*
+ * 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.diag;
+
+import java.io.Serializable;
+import java.util.Map;
+
+/**
+ * Base class for internally emitted events used for diagnostics and testing.
+ */
+public abstract class DiagnosticEvent
+{
+    /**
+     * Event creation time.
+     */
+    public final long timestamp = System.currentTimeMillis();
+
+    /**
+     * Name of allocating thread.
+     */
+    public final String threadName = Thread.currentThread().getName();
+
+    /**
+     * Returns event type discriminator. This will usually be a enum value.
+     */
+    public abstract Enum<?> getType();
+
+    /**
+     * Returns map of key-value pairs containing relevant event details. Values can be complex objects like other
+     * maps, but must be Serializable, as returned values may be consumed by external clients. It's strongly recommended
+     * to stick to standard Java classes to avoid distributing custom classes to clients and also prevent potential
+     * class versioning conflicts.
+     */
+    public abstract Map<String, Serializable> toMap();
+}

src/java/org/apache/cassandra/diag/DiagnosticEventService.java

@@ -0,0 +1,291 @@
+/*
+ * 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.diag;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import com.google.common.collect.ImmutableCollection;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMultimap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableSetMultimap;
+import com.google.common.collect.Iterables;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+
+/**
+ * Service for publishing and consuming {@link DiagnosticEvent}s.
+ */
+public final class DiagnosticEventService
+{
+    private static final Logger logger = LoggerFactory.getLogger(DiagnosticEventService.class);
+
+    // Subscribers interested in consuming all kind of events
+    private ImmutableSet<Consumer<DiagnosticEvent>> subscribersAll = ImmutableSet.of();
+
+    // Subscribers for particular event class, e.g. BootstrapEvent
+    private ImmutableSetMultimap<Class<? extends DiagnosticEvent>, Consumer<DiagnosticEvent>> subscribersByClass = ImmutableSetMultimap.of();
+
+    // Subscribers for event class and type, e.g. BootstrapEvent#TOKENS_ALLOCATED
+    private ImmutableMap<Class, ImmutableSetMultimap<Enum<?>, Consumer<DiagnosticEvent>>> subscribersByClassAndType = ImmutableMap.of();
+
+    private static final DiagnosticEventService instance = new DiagnosticEventService();
+
+    private DiagnosticEventService()
+    {
+    }
+
+    /**
+     * Makes provided event available to all subscribers.
+     */
+    public void publish(DiagnosticEvent event)
+    {
+        if (!DatabaseDescriptor.diagnosticEventsEnabled())
+            return;
+
+        logger.trace("Publishing: {}", event);
+
+        // event class + type
+        ImmutableMultimap<Enum<?>, Consumer<DiagnosticEvent>> consumersByType = subscribersByClassAndType.get(event.getClass());
+        if (consumersByType != null)
+        {
+            ImmutableCollection<Consumer<DiagnosticEvent>> consumers = consumersByType.get(event.getType());
+            if (consumers != null)
+            {
+                for (Consumer<DiagnosticEvent> consumer : consumers)
+                    consumer.accept(event);
+            }
+        }
+
+        // event class
+        Set<Consumer<DiagnosticEvent>> consumersByEvents = subscribersByClass.get(event.getClass());
+        if (consumersByEvents != null)
+        {
+            for (Consumer<DiagnosticEvent> consumer : consumersByEvents)
+                consumer.accept(event);
+        }
+
+        // all events
+        for (Consumer<DiagnosticEvent> consumer : subscribersAll)
+            consumer.accept(event);
+    }
+
+    /**
+     * Registers event handler for specified class of events.
+     * @param event DiagnosticEvent class implementation
+     * @param consumer Consumer for received events
+     */
+    public synchronized <E extends DiagnosticEvent> void subscribe(Class<E> event, Consumer<E> consumer)
+    {
+        subscribersByClass = ImmutableSetMultimap.<Class<? extends DiagnosticEvent>, Consumer<DiagnosticEvent>>builder()
+                              .putAll(subscribersByClass)
+                              .put(event, new TypedConsumerWrapper<>(consumer))
+                              .build();
+    }
+
+    /**
+     * Registers event handler for specified class of events.
+     * @param event DiagnosticEvent class implementation
+     * @param consumer Consumer for received events
+     */
+    public synchronized <E extends DiagnosticEvent, T extends Enum<T>> void subscribe(Class<E> event,
+                                                                                      T eventType,
+                                                                                      Consumer<E> consumer)
+    {
+        ImmutableSetMultimap.Builder<Enum<?>, Consumer<DiagnosticEvent>> byTypeBuilder = ImmutableSetMultimap.builder();
+        if (subscribersByClassAndType.containsKey(event))
+            byTypeBuilder.putAll(subscribersByClassAndType.get(event));
+        byTypeBuilder.put(eventType, new TypedConsumerWrapper<>(consumer));
+
+        ImmutableMap.Builder<Class, ImmutableSetMultimap<Enum<?>, Consumer<DiagnosticEvent>>> byClassBuilder = ImmutableMap.builder();
+        for (Class clazz : subscribersByClassAndType.keySet())
+        {
+            if (!clazz.equals(event))
+                byClassBuilder.put(clazz, subscribersByClassAndType.get(clazz));
+        }
+
+        subscribersByClassAndType = byClassBuilder
+                                    .put(event, byTypeBuilder.build())
+                                    .build();
+    }
+
+    /**
+     * Registers event handler for all DiagnosticEvents published from this point.
+     * @param consumer Consumer for received events
+     */
+    public synchronized void subscribeAll(Consumer<DiagnosticEvent> consumer)
+    {
+        subscribersAll = ImmutableSet.<Consumer<DiagnosticEvent>>builder()
+                         .addAll(subscribersAll)
+                         .add(consumer)
+                         .build();
+    }
+
+    /**
+     * De-registers event handler from receiving any further events.
+     * @param consumer Consumer registered for receiving events
+     */
+    public synchronized <E extends DiagnosticEvent> void unsubscribe(Consumer<E> consumer)
+    {
+        // all events
+        subscribersAll = ImmutableSet.copyOf(Iterables.filter(subscribersAll, (c) -> c != consumer));
+
+        // event class
+        ImmutableSetMultimap.Builder<Class<? extends DiagnosticEvent>, Consumer<DiagnosticEvent>> byClassBuilder = ImmutableSetMultimap.builder();
+        Collection<Map.Entry<Class<? extends DiagnosticEvent>, Consumer<DiagnosticEvent>>> entries = subscribersByClass.entries();
+        for (Map.Entry<Class<? extends DiagnosticEvent>, Consumer<DiagnosticEvent>> entry : entries)
+        {
+            Consumer<DiagnosticEvent> subscriber = entry.getValue();
+            if (subscriber instanceof TypedConsumerWrapper)
+                subscriber = ((TypedConsumerWrapper)subscriber).wrapped;
+
+            if (subscriber != consumer)
+            {
+                byClassBuilder = byClassBuilder.put(entry);
+            }
+        }
+        subscribersByClass = byClassBuilder.build();
+
+
+        // event class + type
+        ImmutableMap.Builder<Class, ImmutableSetMultimap<Enum<?>, Consumer<DiagnosticEvent>>> byClassAndTypeBuilder = ImmutableMap.builder();
+        for (Map.Entry<Class, ImmutableSetMultimap<Enum<?>, Consumer<DiagnosticEvent>>> byClassEntry : subscribersByClassAndType.entrySet())
+        {
+            ImmutableSetMultimap.Builder<Enum<?>, Consumer<DiagnosticEvent>> byTypeBuilder = ImmutableSetMultimap.builder();
+            ImmutableSetMultimap<Enum<?>, Consumer<DiagnosticEvent>> byTypeConsumers = byClassEntry.getValue();
+            Iterables.filter(byTypeConsumers.entries(), (e) ->
+            {
+                if (e == null || e.getValue() == null) return false;
+                Consumer<DiagnosticEvent> subscriber = e.getValue();
+                if (subscriber instanceof TypedConsumerWrapper)
+                    subscriber = ((TypedConsumerWrapper) subscriber).wrapped;
+                return subscriber != consumer;
+            }).forEach(byTypeBuilder::put);
+
+            ImmutableSetMultimap<Enum<?>, Consumer<DiagnosticEvent>> byType = byTypeBuilder.build();
+            if (!byType.isEmpty())
+                byClassAndTypeBuilder.put(byClassEntry.getKey(), byType);
+        }
+
+        subscribersByClassAndType = byClassAndTypeBuilder.build();
+    }
+
+    /**
+     * Indicates if any {@link Consumer} has been registered for the specified class of events.
+     * @param event DiagnosticEvent class implementation
+     */
+    public <E extends DiagnosticEvent> boolean hasSubscribers(Class<E> event)
+    {
+        return !subscribersAll.isEmpty() || subscribersByClass.containsKey(event) || subscribersByClassAndType.containsKey(event);
+    }
+
+    /**
+     * Indicates if any {@link Consumer} has been registered for the specified class of events.
+     * @param event DiagnosticEvent class implementation
+     * @param eventType Subscribed event type matched against {@link DiagnosticEvent#getType()}
+     */
+    public <E extends DiagnosticEvent, T extends Enum<T>> boolean hasSubscribers(Class<E> event, T eventType)
+    {
+        if (!subscribersAll.isEmpty())
+            return true;
+
+        ImmutableSet<Consumer<DiagnosticEvent>> subscribers = subscribersByClass.get(event);
+        if (subscribers != null && !subscribers.isEmpty())
+            return true;
+
+        ImmutableSetMultimap<Enum<?>, Consumer<DiagnosticEvent>> byType = subscribersByClassAndType.get(event);
+        if (byType == null || byType.isEmpty()) return false;
+
+        Set<Consumer<DiagnosticEvent>> consumers = byType.get(eventType);
+        return consumers != null && !consumers.isEmpty();
+    }
+
+    /**
+     * Indicates if events are enabled for specified event class based on {@link DatabaseDescriptor#diagnosticEventsEnabled()}
+     * and {@link #hasSubscribers(Class)}.
+     * @param event DiagnosticEvent class implementation
+     */
+    public <E extends DiagnosticEvent> boolean isEnabled(Class<E> event)
+    {
+        return DatabaseDescriptor.diagnosticEventsEnabled() && hasSubscribers(event);
+    }
+
+    /**
+     * Indicates if events are enabled for specified event class based on {@link DatabaseDescriptor#diagnosticEventsEnabled()}
+     * and {@link #hasSubscribers(Class, Enum)}.
+     * @param event DiagnosticEvent class implementation
+     * @param eventType Subscribed event type matched against {@link DiagnosticEvent#getType()}
+     */
+    public <E extends DiagnosticEvent, T extends Enum<T>> boolean isEnabled(Class<E> event, T eventType)
+    {
+        return DatabaseDescriptor.diagnosticEventsEnabled() && hasSubscribers(event, eventType);
+    }
+
+    public static DiagnosticEventService instance()
+    {
+        return instance;
+    }
+
+    /**
+     * Removes all active subscribers. Should only be called from testing.
+     */
+    public synchronized void cleanup()
+    {
+        subscribersByClass = ImmutableSetMultimap.of();
+        subscribersAll = ImmutableSet.of();
+        subscribersByClassAndType = ImmutableMap.of();
+    }
+
+    /**
+     * Wrapper class for supporting typed event handling for consumers.
+     */
+    private static class TypedConsumerWrapper<E> implements Consumer<DiagnosticEvent>
+    {
+        private final Consumer<E> wrapped;
+
+        private TypedConsumerWrapper(Consumer<E> wrapped)
+        {
+            this.wrapped = wrapped;
+        }
+
+        public void accept(DiagnosticEvent e)
+        {
+            wrapped.accept((E)e);
+        }
+
+        public boolean equals(Object o)
+        {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            TypedConsumerWrapper<?> that = (TypedConsumerWrapper<?>) o;
+            return Objects.equals(wrapped, that.wrapped);
+        }
+
+        public int hashCode()
+        {
+            return Objects.hash(wrapped);
+        }
+    }
+}

src/java/org/apache/cassandra/gms/Gossiper.java

@@ -381,6 +381,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         {
             markDead(endpoint, epState);
         }
+
+        GossiperDiagnostics.convicted(this, endpoint, phi);
     }
 
     /**
@@ -398,6 +400,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         epState.getHeartBeatState().forceHighestPossibleVersionUnsafe();
         markDead(endpoint, epState);
         FailureDetector.instance.forceConviction(endpoint);
+        GossiperDiagnostics.markedAsShutdown(this, endpoint);
     }
 
     /**
@@ -428,6 +431,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         quarantineEndpoint(endpoint);
         if (logger.isDebugEnabled())
             logger.debug("evicting {} from gossip", endpoint);
+        GossiperDiagnostics.evictedFromMembership(this, endpoint);
     }
 
     /**
@@ -453,6 +457,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         MessagingService.instance().destroyConnectionPool(endpoint);
         if (logger.isDebugEnabled())
             logger.debug("removing endpoint {}", endpoint);
+        GossiperDiagnostics.removedEndpoint(this, endpoint);
     }
 
     /**
@@ -474,6 +479,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
     private void quarantineEndpoint(InetAddressAndPort endpoint, long quarantineExpiration)
     {
         justRemovedEndpoints.put(endpoint, quarantineExpiration);
+        GossiperDiagnostics.quarantinedEndpoint(this, endpoint, quarantineExpiration);
     }
 
     /**
@@ -485,6 +491,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         // remember, quarantineEndpoint will effectively already add QUARANTINE_DELAY, so this is 2x
         logger.debug("");
         quarantineEndpoint(endpoint, System.currentTimeMillis() + QUARANTINE_DELAY);
+        GossiperDiagnostics.replacementQuarantine(this, endpoint);
     }
 
     /**
@@ -498,6 +505,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         removeEndpoint(endpoint);
         evictFromMembership(endpoint);
         replacementQuarantine(endpoint);
+        GossiperDiagnostics.replacedEndpoint(this, endpoint);
     }
 
     /**
@@ -688,7 +696,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         if (firstSynSendAt == 0)
             firstSynSendAt = System.nanoTime();
         MessagingService.instance().sendOneWay(message, to);
-        return seeds.contains(to);
+
+        boolean isSeed = seeds.contains(to);
+        GossiperDiagnostics.sendGossipDigestSyn(this, to);
+        return isSeed;
     }
 
     /* Sends a Gossip message to a live member and returns true if the recipient was a seed */
@@ -889,6 +900,31 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         return endpointStateMap.size();
     }
 
+    Map<InetAddressAndPort, EndpointState> getEndpointStateMap()
+    {
+        return ImmutableMap.copyOf(endpointStateMap);
+    }
+
+    Map<InetAddressAndPort, Long> getJustRemovedEndpoints()
+    {
+        return ImmutableMap.copyOf(justRemovedEndpoints);
+    }
+
+    Map<InetAddressAndPort, Long> getUnreachableEndpoints()
+    {
+        return ImmutableMap.copyOf(unreachableEndpoints);
+    }
+
+    Set<InetAddressAndPort> getSeedsInShadowRound()
+    {
+        return ImmutableSet.copyOf(seedsInShadowRound);
+    }
+
+    long getLastProcessedMessageAt()
+    {
+        return lastProcessedMessageAt;
+    }
+
     public UUID getHostId(InetAddressAndPort endpoint)
     {
         return getHostId(endpoint, endpointStateMap);
@@ -1028,6 +1064,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         };
 
         MessagingService.instance().sendRR(echoMessage, addr, echoHandler);
+
+        GossiperDiagnostics.markedAlive(this, addr, localState);
     }
 
     @VisibleForTesting
@@ -1046,6 +1084,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
             subscriber.onAlive(addr, localState);
         if (logger.isTraceEnabled())
             logger.trace("Notified {}", subscribers);
+
+        GossiperDiagnostics.realMarkedAlive(this, addr, localState);
     }
 
     @VisibleForTesting
@@ -1061,6 +1101,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
             subscriber.onDead(addr, localState);
         if (logger.isTraceEnabled())
             logger.trace("Notified {}", subscribers);
+
+        GossiperDiagnostics.markedDead(this, addr, localState);
     }
 
     /**
@@ -1101,6 +1143,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         // check this at the end so nodes will learn about the endpoint
         if (isShutdown(ep))
             markAsShutdown(ep);
+
+        GossiperDiagnostics.majorStateChangeHandled(this, ep, epState);
     }
 
     public boolean isAlive(InetAddressAndPort endpoint)

src/java/org/apache/cassandra/gms/GossiperDiagnostics.java

@@ -0,0 +1,113 @@
+/*
+ * 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.gms;
+
+
+import org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.gms.GossiperEvent.GossiperEventType;
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+/**
+ * Utility methods for DiagnosticEvent activities.
+ */
+final class GossiperDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private GossiperDiagnostics()
+    {
+    }
+
+    static void markedAsShutdown(Gossiper gossiper, InetAddressAndPort endpoint)
+    {
+        if (isEnabled(GossiperEventType.MARKED_AS_SHUTDOWN))
+            service.publish(new GossiperEvent(GossiperEventType.MARKED_AS_SHUTDOWN, gossiper, endpoint, null, null));
+    }
+
+    static void convicted(Gossiper gossiper, InetAddressAndPort endpoint, double phi)
+    {
+        if (isEnabled(GossiperEventType.CONVICTED))
+            service.publish(new GossiperEvent(GossiperEventType.CONVICTED, gossiper, endpoint, null, null));
+    }
+
+    static void replacementQuarantine(Gossiper gossiper, InetAddressAndPort endpoint)
+    {
+        if (isEnabled(GossiperEventType.REPLACEMENT_QUARANTINE))
+            service.publish(new GossiperEvent(GossiperEventType.REPLACEMENT_QUARANTINE, gossiper, endpoint, null, null));
+    }
+
+    static void replacedEndpoint(Gossiper gossiper, InetAddressAndPort endpoint)
+    {
+        if (isEnabled(GossiperEventType.REPLACED_ENDPOINT))
+            service.publish(new GossiperEvent(GossiperEventType.REPLACED_ENDPOINT, gossiper, endpoint, null, null));
+    }
+
+    static void evictedFromMembership(Gossiper gossiper, InetAddressAndPort endpoint)
+    {
+        if (isEnabled(GossiperEventType.EVICTED_FROM_MEMBERSHIP))
+            service.publish(new GossiperEvent(GossiperEventType.EVICTED_FROM_MEMBERSHIP, gossiper, endpoint, null, null));
+    }
+
+    static void removedEndpoint(Gossiper gossiper, InetAddressAndPort endpoint)
+    {
+        if (isEnabled(GossiperEventType.REMOVED_ENDPOINT))
+            service.publish(new GossiperEvent(GossiperEventType.REMOVED_ENDPOINT, gossiper, endpoint, null, null));
+    }
+
+    static void quarantinedEndpoint(Gossiper gossiper, InetAddressAndPort endpoint, long quarantineExpiration)
+    {
+        if (isEnabled(GossiperEventType.QUARANTINED_ENDPOINT))
+            service.publish(new GossiperEvent(GossiperEventType.QUARANTINED_ENDPOINT, gossiper, endpoint, quarantineExpiration, null));
+    }
+
+    static void markedAlive(Gossiper gossiper, InetAddressAndPort addr, EndpointState localState)
+    {
+        if (isEnabled(GossiperEventType.MARKED_ALIVE))
+            service.publish(new GossiperEvent(GossiperEventType.MARKED_ALIVE, gossiper, addr, null, localState));
+    }
+
+    static void realMarkedAlive(Gossiper gossiper, InetAddressAndPort addr, EndpointState localState)
+    {
+        if (isEnabled(GossiperEventType.REAL_MARKED_ALIVE))
+            service.publish(new GossiperEvent(GossiperEventType.REAL_MARKED_ALIVE, gossiper, addr, null, localState));
+    }
+
+    static void markedDead(Gossiper gossiper, InetAddressAndPort addr, EndpointState localState)
+    {
+        if (isEnabled(GossiperEventType.MARKED_DEAD))
+            service.publish(new GossiperEvent(GossiperEventType.MARKED_DEAD, gossiper, addr, null, localState));
+    }
+
+    static void majorStateChangeHandled(Gossiper gossiper, InetAddressAndPort addr, EndpointState state)
+    {
+        if (isEnabled(GossiperEventType.MAJOR_STATE_CHANGE_HANDLED))
+            service.publish(new GossiperEvent(GossiperEventType.MAJOR_STATE_CHANGE_HANDLED, gossiper, addr, null, state));
+    }
+
+    static void sendGossipDigestSyn(Gossiper gossiper, InetAddressAndPort to)
+    {
+        if (isEnabled(GossiperEventType.SEND_GOSSIP_DIGEST_SYN))
+            service.publish(new GossiperEvent(GossiperEventType.SEND_GOSSIP_DIGEST_SYN, gossiper, to, null, null));
+    }
+
+    private static boolean isEnabled(GossiperEventType type)
+    {
+        return service.isEnabled(GossiperEvent.class, type);
+    }
+}

src/java/org/apache/cassandra/gms/GossiperEvent.java

@@ -0,0 +1,111 @@
+/*
+ * 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.gms;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.diag.DiagnosticEvent;
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+/**
+ * DiagnosticEvent implementation for {@link Gossiper} activities.
+ */
+final class GossiperEvent extends DiagnosticEvent
+{
+    private final InetAddressAndPort endpoint;
+    @Nullable
+    private final Long quarantineExpiration;
+    @Nullable
+    private final EndpointState localState;
+
+    private final Map<InetAddressAndPort, EndpointState> endpointStateMap;
+    private final boolean inShadowRound;
+    private final Map<InetAddressAndPort, Long> justRemovedEndpoints;
+    private final long lastProcessedMessageAt;
+    private final Set<InetAddressAndPort> liveEndpoints;
+    private final List<String> seeds;
+    private final Set<InetAddressAndPort> seedsInShadowRound;
+    private final Map<InetAddressAndPort, Long> unreachableEndpoints;
+
+
+    enum GossiperEventType
+    {
+        MARKED_AS_SHUTDOWN,
+        CONVICTED,
+        REPLACEMENT_QUARANTINE,
+        REPLACED_ENDPOINT,
+        EVICTED_FROM_MEMBERSHIP,
+        REMOVED_ENDPOINT,
+        QUARANTINED_ENDPOINT,
+        MARKED_ALIVE,
+        REAL_MARKED_ALIVE,
+        MARKED_DEAD,
+        MAJOR_STATE_CHANGE_HANDLED,
+        SEND_GOSSIP_DIGEST_SYN
+    }
+
+    public GossiperEventType type;
+
+
+    GossiperEvent(GossiperEventType type, Gossiper gossiper, InetAddressAndPort endpoint,
+                  @Nullable Long quarantineExpiration, @Nullable EndpointState localState)
+    {
+        this.type = type;
+        this.endpoint = endpoint;
+        this.quarantineExpiration = quarantineExpiration;
+        this.localState = localState;
+
+        this.endpointStateMap = gossiper.getEndpointStateMap();
+        this.inShadowRound = gossiper.isInShadowRound();
+        this.justRemovedEndpoints = gossiper.getJustRemovedEndpoints();
+        this.lastProcessedMessageAt = gossiper.getLastProcessedMessageAt();
+        this.liveEndpoints = gossiper.getLiveMembers();
+        this.seeds = gossiper.getSeeds();
+        this.seedsInShadowRound = gossiper.getSeedsInShadowRound();
+        this.unreachableEndpoints = gossiper.getUnreachableEndpoints();
+    }
+
+    public Enum<GossiperEventType> getType()
+    {
+        return type;
+    }
+
+    public HashMap<String, Serializable> toMap()
+    {
+        // be extra defensive against nulls and bugs
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (endpoint != null) ret.put("endpoint", endpoint.getHostAddress(true));
+        ret.put("quarantineExpiration", quarantineExpiration);
+        ret.put("localState", String.valueOf(localState));
+        ret.put("endpointStateMap", String.valueOf(endpointStateMap));
+        ret.put("inShadowRound", inShadowRound);
+        ret.put("justRemovedEndpoints", String.valueOf(justRemovedEndpoints));
+        ret.put("lastProcessedMessageAt", lastProcessedMessageAt);
+        ret.put("liveEndpoints", String.valueOf(liveEndpoints));
+        ret.put("seeds", String.valueOf(seeds));
+        ret.put("seedsInShadowRound", String.valueOf(seedsInShadowRound));
+        ret.put("unreachableEndpoints", String.valueOf(unreachableEndpoints));
+        return ret;
+    }
+}

src/java/org/apache/cassandra/hints/Hint.java

@@ -132,7 +132,7 @@ public final class Hint
     /**
      * @return calculates whether or not it is safe to apply the hint without risking to resurrect any deleted data
      */
-    boolean isLive()
+    public boolean isLive()
     {
         return isLive(creationTime, System.currentTimeMillis(), ttl());
     }

src/java/org/apache/cassandra/hints/HintDiagnostics.java

@@ -0,0 +1,85 @@
+/*
+ * 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.hints;
+
+import org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.hints.HintEvent.HintEventType;
+import org.apache.cassandra.hints.HintEvent.HintResult;
+
+/**
+ * Utility methods for DiagnosticEvents around hinted handoff.
+ */
+final class HintDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private HintDiagnostics()
+    {
+    }
+
+    static void dispatcherCreated(HintsDispatcher dispatcher)
+    {
+        if (isEnabled(HintEventType.DISPATCHER_CREATED))
+            service.publish(new HintEvent(HintEventType.DISPATCHER_CREATED, dispatcher,
+                                          dispatcher.hostId, dispatcher.address, null, null, null, null));
+    }
+
+    static void dispatcherClosed(HintsDispatcher dispatcher)
+    {
+        if (isEnabled(HintEventType.DISPATCHER_CLOSED))
+            service.publish(new HintEvent(HintEventType.DISPATCHER_CLOSED, dispatcher,
+                                          dispatcher.hostId, dispatcher.address, null, null, null, null));
+    }
+
+    static void dispatchPage(HintsDispatcher dispatcher)
+    {
+        if (isEnabled(HintEventType.DISPATCHER_PAGE))
+            service.publish(new HintEvent(HintEventType.DISPATCHER_PAGE, dispatcher,
+                                          dispatcher.hostId, dispatcher.address, null, null, null, null));
+    }
+
+    static void abortRequested(HintsDispatcher dispatcher)
+    {
+        if (isEnabled(HintEventType.ABORT_REQUESTED))
+            service.publish(new HintEvent(HintEventType.ABORT_REQUESTED, dispatcher,
+                                          dispatcher.hostId, dispatcher.address, null, null, null, null));
+    }
+
+    static void pageSuccessResult(HintsDispatcher dispatcher, long success, long failures, long timeouts)
+    {
+        if (isEnabled(HintEventType.DISPATCHER_HINT_RESULT))
+            service.publish(new HintEvent(HintEventType.DISPATCHER_HINT_RESULT, dispatcher,
+                                          dispatcher.hostId, dispatcher.address, HintResult.PAGE_SUCCESS,
+                                          success, failures, timeouts));
+    }
+
+    static void pageFailureResult(HintsDispatcher dispatcher, long success, long failures, long timeouts)
+    {
+        if (isEnabled(HintEventType.DISPATCHER_HINT_RESULT))
+            service.publish(new HintEvent(HintEventType.DISPATCHER_HINT_RESULT, dispatcher,
+                                          dispatcher.hostId, dispatcher.address, HintResult.PAGE_FAILURE,
+                                          success, failures, timeouts));
+    }
+
+    private static boolean isEnabled(HintEventType type)
+    {
+        return service.isEnabled(HintEvent.class, type);
+    }
+
+}

src/java/org/apache/cassandra/hints/HintEvent.java

@@ -0,0 +1,102 @@
+/*
+ * 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.hints;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.UUID;
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.diag.DiagnosticEvent;
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+/**
+ * DiagnosticEvent implementation for hinted handoff.
+ */
+final class HintEvent extends DiagnosticEvent
+{
+    enum HintEventType
+    {
+        DISPATCHING_STARTED,
+        DISPATCHING_PAUSED,
+        DISPATCHING_RESUMED,
+        DISPATCHING_SHUTDOWN,
+
+        DISPATCHER_CREATED,
+        DISPATCHER_CLOSED,
+
+        DISPATCHER_PAGE,
+        DISPATCHER_HINT_RESULT,
+
+        ABORT_REQUESTED
+    }
+
+    enum HintResult
+    {
+        PAGE_SUCCESS, PAGE_FAILURE
+    }
+
+    private final HintEventType type;
+    private final HintsDispatcher dispatcher;
+    private final UUID targetHostId;
+    private final InetAddressAndPort targetAddress;
+    @Nullable
+    private final HintResult dispatchResult;
+    @Nullable
+    private final Long pageHintsSuccessful;
+    @Nullable
+    private final Long pageHintsFailed;
+    @Nullable
+    private final Long pageHintsTimeout;
+
+    HintEvent(HintEventType type, HintsDispatcher dispatcher, UUID targetHostId, InetAddressAndPort targetAddress,
+              @Nullable HintResult dispatchResult, @Nullable Long pageHintsSuccessful,
+              @Nullable Long pageHintsFailed, @Nullable Long pageHintsTimeout)
+    {
+        this.type = type;
+        this.dispatcher = dispatcher;
+        this.targetHostId = targetHostId;
+        this.targetAddress = targetAddress;
+        this.dispatchResult = dispatchResult;
+        this.pageHintsSuccessful = pageHintsSuccessful;
+        this.pageHintsFailed = pageHintsFailed;
+        this.pageHintsTimeout = pageHintsTimeout;
+    }
+
+    public Enum<HintEventType> getType()
+    {
+        return type;
+    }
+
+    public HashMap<String, Serializable> toMap()
+    {
+        // be extra defensive against nulls and bugs
+        HashMap<String, Serializable> ret = new HashMap<>();
+        ret.put("targetHostId", targetHostId);
+        ret.put("targetAddress", targetAddress.getHostAddress(true));
+        if (dispatchResult != null) ret.put("dispatchResult", dispatchResult.name());
+        if (pageHintsSuccessful != null || pageHintsFailed != null || pageHintsTimeout != null)
+        {
+            ret.put("hint.page.hints_succeeded", pageHintsSuccessful);
+            ret.put("hint.page.hints_failed", pageHintsFailed);
+            ret.put("hint.page.hints_timed_out", pageHintsTimeout);
+        }
+        return ret;
+    }
+}

src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java

@@ -306,4 +306,14 @@ final class HintsDispatchExecutor
             }
         }
     }
+
+    public boolean isPaused()
+    {
+        return isPaused.get();
+    }
+
+    public boolean hasScheduledDispatches()
+    {
+        return !scheduledDispatches.isEmpty();
+    }
 }

src/java/org/apache/cassandra/hints/HintsDispatcher.java

@@ -50,8 +50,8 @@ final class HintsDispatcher implements AutoCloseable
     private enum Action { CONTINUE, ABORT }
 
     private final HintsReader reader;
-    private final UUID hostId;
-    private final InetAddressAndPort address;
+    final UUID hostId;
+    final InetAddressAndPort address;
     private final int messagingVersion;
     private final BooleanSupplier abortRequested;
 
@@ -71,11 +71,14 @@ final class HintsDispatcher implements AutoCloseable
     static HintsDispatcher create(File file, RateLimiter rateLimiter, InetAddressAndPort address, UUID hostId, BooleanSupplier abortRequested)
     {
         int messagingVersion = MessagingService.instance().getVersion(address);
-        return new HintsDispatcher(HintsReader.open(file, rateLimiter), hostId, address, messagingVersion, abortRequested);
+        HintsDispatcher dispatcher = new HintsDispatcher(HintsReader.open(file, rateLimiter), hostId, address, messagingVersion, abortRequested);
+        HintDiagnostics.dispatcherCreated(dispatcher);
+        return dispatcher;
     }
 
     public void close()
     {
+        HintDiagnostics.dispatcherClosed(this);
         reader.close();
     }
 
@@ -111,6 +114,7 @@ final class HintsDispatcher implements AutoCloseable
     // retry in case of a timeout; stop in case of a failure, host going down, or delivery paused
     private Action dispatch(HintsReader.Page page)
     {
+        HintDiagnostics.dispatchPage(this);
         return sendHintsAndAwait(page);
     }
 
@@ -132,33 +136,34 @@ final class HintsDispatcher implements AutoCloseable
         if (action == Action.ABORT)
             return action;
 
-        boolean hadFailures = false;
+        long success = 0, failures = 0, timeouts = 0;
         for (Callback cb : callbacks)
         {
             Callback.Outcome outcome = cb.await();
-            updateMetrics(outcome);
-
-            if (outcome != Callback.Outcome.SUCCESS)
-                hadFailures = true;
+            if (outcome == Callback.Outcome.SUCCESS) success++;
+            else if (outcome == Callback.Outcome.FAILURE) failures++;
+            else if (outcome == Callback.Outcome.TIMEOUT) timeouts++;
         }
 
-        return hadFailures ? Action.ABORT : Action.CONTINUE;
+        updateMetrics(success, failures, timeouts);
+
+        if (failures > 0 || timeouts > 0)
+        {
+            HintDiagnostics.pageFailureResult(this, success, failures, timeouts);
+            return Action.ABORT;
+        }
+        else
+        {
+            HintDiagnostics.pageSuccessResult(this, success, failures, timeouts);
+            return Action.CONTINUE;
+        }
     }
 
-    private void updateMetrics(Callback.Outcome outcome)
+    private void updateMetrics(long success, long failures, long timeouts)
     {
-        switch (outcome)
-        {
-            case SUCCESS:
-                HintsServiceMetrics.hintsSucceeded.mark();
-                break;
-            case FAILURE:
-                HintsServiceMetrics.hintsFailed.mark();
-                break;
-            case TIMEOUT:
-                HintsServiceMetrics.hintsTimedOut.mark();
-                break;
-        }
+        HintsServiceMetrics.hintsSucceeded.mark(success);
+        HintsServiceMetrics.hintsFailed.mark(failures);
+        HintsServiceMetrics.hintsTimedOut.mark(timeouts);
     }
 
     /*
@@ -170,7 +175,10 @@ final class HintsDispatcher implements AutoCloseable
         while (hints.hasNext())
         {
             if (abortRequested.getAsBoolean())
+            {
+                HintDiagnostics.abortRequested(this);
                 return Action.ABORT;
+            }
             callbacks.add(sendFunction.apply(hints.next()));
         }
         return Action.CONTINUE;

src/java/org/apache/cassandra/hints/HintsService.java

@@ -73,8 +73,8 @@ public final class HintsService implements HintsServiceMBean
     private final HintsCatalog catalog;
     private final HintsWriteExecutor writeExecutor;
     private final HintsBufferPool bufferPool;
-    private final HintsDispatchExecutor dispatchExecutor;
-    private final AtomicBoolean isDispatchPaused;
+    final HintsDispatchExecutor dispatchExecutor;
+    final AtomicBoolean isDispatchPaused;
 
     private volatile boolean isShutDown = false;
 
@@ -209,6 +209,8 @@ public final class HintsService implements HintsServiceMBean
 
         isDispatchPaused.set(false);
 
+        HintsServiceDiagnostics.dispatchingStarted(this);
+
         HintsDispatchTrigger trigger = new HintsDispatchTrigger(catalog, writeExecutor, dispatchExecutor, isDispatchPaused);
         // triggering hint dispatch is now very cheap, so we can do it more often - every 10 seconds vs. every 10 minutes,
         // previously; this reduces mean time to delivery, and positively affects batchlog delivery latencies, too
@@ -219,12 +221,16 @@ public final class HintsService implements HintsServiceMBean
     {
         logger.info("Paused hints dispatch");
         isDispatchPaused.set(true);
+
+        HintsServiceDiagnostics.dispatchingPaused(this);
     }
 
     public void resumeDispatch()
     {
         logger.info("Resumed hints dispatch");
         isDispatchPaused.set(false);
+
+        HintsServiceDiagnostics.dispatchingResumed(this);
     }
 
     /**
@@ -250,6 +256,8 @@ public final class HintsService implements HintsServiceMBean
 
         dispatchExecutor.shutdownBlocking();
         writeExecutor.shutdownBlocking();
+
+        HintsServiceDiagnostics.dispatchingShutdown(this);
     }
 
     /**

src/java/org/apache/cassandra/hints/HintsServiceDiagnostics.java

@@ -0,0 +1,65 @@
+/*
+ * 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.hints;
+
+
+import org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.hints.HintsServiceEvent.HintsServiceEventType;
+
+/**
+ * Utility methods for DiagnosticEvents around the HintService.
+ */
+final class HintsServiceDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private HintsServiceDiagnostics()
+    {
+    }
+    
+    static void dispatchingStarted(HintsService hintsService)
+    {
+        if (isEnabled(HintsServiceEventType.DISPATCHING_STARTED))
+            service.publish(new HintsServiceEvent(HintsServiceEventType.DISPATCHING_STARTED, hintsService));
+    }
+
+    static void dispatchingShutdown(HintsService hintsService)
+    {
+        if (isEnabled(HintsServiceEventType.DISPATCHING_SHUTDOWN))
+            service.publish(new HintsServiceEvent(HintsServiceEventType.DISPATCHING_SHUTDOWN, hintsService));
+    }
+
+    static void dispatchingPaused(HintsService hintsService)
+    {
+        if (isEnabled(HintsServiceEventType.DISPATCHING_PAUSED))
+            service.publish(new HintsServiceEvent(HintsServiceEventType.DISPATCHING_PAUSED, hintsService));
+    }
+
+    static void dispatchingResumed(HintsService hintsService)
+    {
+        if (isEnabled(HintsServiceEventType.DISPATCHING_RESUMED))
+            service.publish(new HintsServiceEvent(HintsServiceEventType.DISPATCHING_RESUMED, hintsService));
+    }
+
+    private static boolean isEnabled(HintsServiceEventType type)
+    {
+        return service.isEnabled(HintsServiceEvent.class, type);
+    }
+
+}

src/java/org/apache/cassandra/hints/HintsServiceEvent.java

@@ -0,0 +1,71 @@
+/*
+ * 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.hints;
+
+import java.io.Serializable;
+import java.util.HashMap;
+
+import org.apache.cassandra.diag.DiagnosticEvent;
+
+/**
+ * DiagnosticEvent implementation for HintService.
+ */
+final class HintsServiceEvent extends DiagnosticEvent
+{
+    enum HintsServiceEventType
+    {
+        DISPATCHING_STARTED,
+        DISPATCHING_PAUSED,
+        DISPATCHING_RESUMED,
+        DISPATCHING_SHUTDOWN
+    }
+
+    private final HintsServiceEventType type;
+    private final HintsService service;
+    private final boolean isDispatchPaused;
+    private final boolean isShutdown;
+    private final boolean dispatchExecutorIsPaused;
+    private final boolean dispatchExecutorHasScheduledDispatches;
+
+    HintsServiceEvent(HintsServiceEventType type, HintsService service)
+    {
+        this.type = type;
+        this.service = service;
+        this.isDispatchPaused = service.isDispatchPaused.get();
+        this.isShutdown = service.isShutDown();
+        this.dispatchExecutorIsPaused = service.dispatchExecutor.isPaused();
+        this.dispatchExecutorHasScheduledDispatches = service.dispatchExecutor.hasScheduledDispatches();
+    }
+
+    public Enum<HintsServiceEventType> getType()
+    {
+        return type;
+    }
+
+    public HashMap<String, Serializable> toMap()
+    {
+        // be extra defensive against nulls and bugs
+        HashMap<String, Serializable> ret = new HashMap<>();
+        ret.put("isDispatchPaused", isDispatchPaused);
+        ret.put("isShutdown", isShutdown);
+        ret.put("dispatchExecutorIsPaused", dispatchExecutorIsPaused);
+        ret.put("dispatchExecutorHasScheduledDispatches", dispatchExecutorHasScheduledDispatches);
+        return ret;
+    }
+}

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

@@ -800,6 +800,8 @@ public class TokenMetadata
         // avoid race between both branches - do not use a lock here as this will block any other unrelated operations!
         synchronized (pendingRanges)
         {
+            TokenMetadataDiagnostics.pendingRangeCalculationStarted(this, keyspaceName);
+
             if (bootstrapTokens.isEmpty() && leavingEndpoints.isEmpty() && movingEndpoints.isEmpty())
             {
                 if (logger.isTraceEnabled())

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

@@ -0,0 +1,46 @@
+/*
+ * 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 org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.locator.TokenMetadataEvent.TokenMetadataEventType;
+
+/**
+ * Utility methods for events related to {@link TokenMetadata} changes.
+ */
+final class TokenMetadataDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private TokenMetadataDiagnostics()
+    {
+    }
+
+    static void pendingRangeCalculationStarted(TokenMetadata tokenMetadata, String keyspace)
+    {
+        if (isEnabled(TokenMetadataEventType.PENDING_RANGE_CALCULATION_STARTED))
+            service.publish(new TokenMetadataEvent(TokenMetadataEventType.PENDING_RANGE_CALCULATION_STARTED, tokenMetadata, keyspace));
+    }
+
+    private static boolean isEnabled(TokenMetadataEventType type)
+    {
+        return service.isEnabled(TokenMetadataEvent.class, type);
+    }
+
+}

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

@@ -0,0 +1,62 @@
+/*
+ * 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.io.Serializable;
+import java.util.HashMap;
+
+import org.apache.cassandra.diag.DiagnosticEvent;
+
+/**
+ * Events related to {@link TokenMetadata} changes.
+ */
+public final class TokenMetadataEvent extends DiagnosticEvent
+{
+
+    public enum TokenMetadataEventType
+    {
+        PENDING_RANGE_CALCULATION_STARTED,
+        PENDING_RANGE_CALCULATION_COMPLETED,
+    }
+
+    private final TokenMetadataEventType type;
+    private final TokenMetadata tokenMetadata;
+    private final String keyspace;
+
+    TokenMetadataEvent(TokenMetadataEventType type, TokenMetadata tokenMetadata, String keyspace)
+    {
+        this.type = type;
+        this.tokenMetadata = tokenMetadata;
+        this.keyspace = keyspace;
+    }
+
+    public TokenMetadataEventType getType()
+    {
+        return type;
+    }
+
+    public HashMap<String, Serializable> toMap()
+    {
+        // be extra defensive against nulls and bugs
+        HashMap<String, Serializable> ret = new HashMap<>();
+        ret.put("keyspace", keyspace);
+        ret.put("tokenMetadata", tokenMetadata.toString());
+        return ret;
+    }
+}

src/java/org/apache/cassandra/schema/Diff.java

@@ -55,5 +55,10 @@ public class Diff<T extends Iterable, S>
             this.after = after;
             this.kind = kind;
         }
+
+        public String toString()
+        {
+            return String.format("%s -> %s (%s)", before, after, kind);
+        }
     }
 }

src/java/org/apache/cassandra/schema/MigrationManager.java

@@ -72,8 +72,9 @@ public class MigrationManager
     {
         if (Schema.instance.getVersion() == null)
         {
-            logger.debug("Not pulling schema from {}, because local schama version is not known yet",
+            logger.debug("Not pulling schema from {}, because local schema version is not known yet",
                          endpoint);
+            SchemaMigrationDiagnostics.unknownLocalSchemaVersion(endpoint, theirVersion);
             return;
         }
         if (Schema.instance.isSameVersion(theirVersion))
@@ -81,12 +82,14 @@ public class MigrationManager
             logger.debug("Not pulling schema from {}, because schema versions match ({})",
                          endpoint,
                          Schema.schemaVersionToString(theirVersion));
+            SchemaMigrationDiagnostics.versionMatch(endpoint, theirVersion);
             return;
         }
         if (!shouldPullSchemaFrom(endpoint))
         {
             logger.debug("Not pulling schema from {}, because versions match ({}/{}), or shouldPullSchemaFrom returned false",
                          endpoint, Schema.instance.getVersion(), theirVersion);
+            SchemaMigrationDiagnostics.skipPull(endpoint, theirVersion);
             return;
         }
 
@@ -319,10 +322,22 @@ public class MigrationManager
     {
         Future<?> f = StageManager.getStage(Stage.MIGRATION).submit(() -> Schema.instance.mergeAndAnnounceVersion(schema));
 
+        Set<InetAddressAndPort> schemaDestinationEndpoints = new HashSet<>();
+        Set<InetAddressAndPort> schemaEndpointsIgnored = new HashSet<>();
         for (InetAddressAndPort endpoint : Gossiper.instance.getLiveMembers())
+        {
             if (shouldPushSchemaTo(endpoint))
+            {
                 pushSchemaMutation(endpoint, schema);
+                schemaDestinationEndpoints.add(endpoint);
+            }
+            else
+            {
+                schemaEndpointsIgnored.add(endpoint);
+            }
+        }
 
+        SchemaAnnouncementDiagnostics.schemaMutationsAnnounced(schemaDestinationEndpoints, schemaEndpointsIgnored);
         FBUtilities.waitOnFuture(f);
     }
 
@@ -340,9 +355,23 @@ public class MigrationManager
         if (locally || result.diff.isEmpty())
             return result.diff;
 
+        Set<InetAddressAndPort> schemaDestinationEndpoints = new HashSet<>();
+        Set<InetAddressAndPort> schemaEndpointsIgnored = new HashSet<>();
         for (InetAddressAndPort endpoint : Gossiper.instance.getLiveMembers())
+        {
             if (shouldPushSchemaTo(endpoint))
+            {
                 pushSchemaMutation(endpoint, result.mutations);
+                schemaDestinationEndpoints.add(endpoint);
+            }
+            else
+            {
+                schemaEndpointsIgnored.add(endpoint);
+            }
+        }
+
+        SchemaAnnouncementDiagnostics.schemaTransformationAnnounced(schemaDestinationEndpoints, schemaEndpointsIgnored,
+                                                                    transformation);
 
         return result.diff;
     }
@@ -357,6 +386,8 @@ public class MigrationManager
 
         logger.debug("Truncating schema tables...");
 
+        SchemaMigrationDiagnostics.resetLocalSchema();
+
         SchemaKeyspace.truncate();
 
         logger.debug("Clearing local schema keyspace definitions...");

src/java/org/apache/cassandra/schema/MigrationTask.java

@@ -51,6 +51,7 @@ final class MigrationTask extends WrappedRunnable
     MigrationTask(InetAddressAndPort endpoint)
     {
         this.endpoint = endpoint;
+        SchemaMigrationDiagnostics.taskCreated(endpoint);
     }
 
     static ConcurrentLinkedQueue<CountDownLatch> getInflightTasks()
@@ -63,6 +64,7 @@ final class MigrationTask extends WrappedRunnable
         if (!FailureDetector.instance.isAlive(endpoint))
         {
             logger.warn("Can't send schema pull request: node {} is down.", endpoint);
+            SchemaMigrationDiagnostics.taskSendAborted(endpoint);
             return;
         }
 
@@ -72,6 +74,7 @@ final class MigrationTask extends WrappedRunnable
         if (!MigrationManager.shouldPullSchemaFrom(endpoint))
         {
             logger.info("Skipped sending a migration request: node {} has a higher major version now.", endpoint);
+            SchemaMigrationDiagnostics.taskSendAborted(endpoint);
             return;
         }
 
@@ -109,5 +112,7 @@ final class MigrationTask extends WrappedRunnable
             inflightTasks.offer(completionLatch);
 
         MessagingService.instance().sendRR(message, endpoint, cb);
+
+        SchemaMigrationDiagnostics.taskRequestSend(endpoint);
     }
 }

src/java/org/apache/cassandra/schema/Schema.java

@@ -98,9 +98,11 @@ public final class Schema
      */
     public void loadFromDisk(boolean updateVersion)
     {
+        SchemaDiagnostics.schemataLoading(this);
         SchemaKeyspace.fetchNonSystemKeyspaces().forEach(this::load);
         if (updateVersion)
             updateVersion();
+        SchemaDiagnostics.schemataLoaded(this);
     }
 
     /**
@@ -128,6 +130,8 @@ public final class Schema
         ksm.tables
            .indexTables()
            .forEach((name, metadata) -> indexMetadataRefs.put(Pair.create(ksm.name, name), new TableMetadataRef(metadata)));
+
+        SchemaDiagnostics.metadataInitialized(this, ksm);
     }
 
     private void reload(KeyspaceMetadata previous, KeyspaceMetadata updated)
@@ -166,6 +170,8 @@ public final class Schema
                    .stream()
                    .map(MapDifference.ValueDifference::rightValue)
                    .forEach(indexTable -> indexMetadataRefs.get(Pair.create(indexTable.keyspace, indexTable.indexName().get())).set(indexTable));
+
+        SchemaDiagnostics.metadataReloaded(this, previous, updated, tablesDiff, viewsDiff, indexesDiff);
     }
 
     public void registerListener(SchemaChangeListener listener)
@@ -249,6 +255,8 @@ public final class Schema
            .indexTables()
            .keySet()
            .forEach(name -> indexMetadataRefs.remove(Pair.create(ksm.name, name)));
+
+        SchemaDiagnostics.metadataRemoved(this, ksm);
     }
 
     public int getNumberOfTables()
@@ -356,6 +364,11 @@ public final class Schema
         return indexMetadataRefs.get(Pair.create(keyspace, index));
     }
 
+    Map<Pair<String, String>, TableMetadataRef> getIndexTableMetadataRefs()
+    {
+        return indexMetadataRefs;
+    }
+
     /**
      * Get Table metadata by its identifier
      *
@@ -373,6 +386,11 @@ public final class Schema
         return getTableMetadataRef(descriptor.ksname, descriptor.cfname);
     }
 
+    Map<TableId, TableMetadataRef> getTableMetadataRefs()
+    {
+        return metadataRefs;
+    }
+
     /**
      * Given a keyspace name and table name, get the table
      * meta data. If the keyspace name or table name is not valid
@@ -511,6 +529,7 @@ public final class Schema
     {
         version = SchemaKeyspace.calculateSchemaDigest();
         SystemKeyspace.updateSchemaVersion(version);
+        SchemaDiagnostics.versionUpdated(this);
     }
 
     /*
@@ -529,6 +548,7 @@ public final class Schema
     private void passiveAnnounceVersion()
     {
         Gossiper.instance.addLocalApplicationState(ApplicationState.SCHEMA, StorageService.instance.valueFactory.schema(version));
+        SchemaDiagnostics.versionAnnounced(this);
     }
 
     /**
@@ -538,6 +558,7 @@ public final class Schema
     {
         getNonSystemKeyspaces().forEach(k -> unload(getKeyspaceMetadata(k)));
         updateVersionAndAnnounce();
+        SchemaDiagnostics.schemataCleared(this);
     }
 
     /*
@@ -646,6 +667,8 @@ public final class Schema
 
     private void alterKeyspace(KeyspaceDiff delta)
     {
+        SchemaDiagnostics.keyspaceAltering(this, delta);
+
         // drop tables and views
         delta.views.dropped.forEach(this::dropView);
         delta.tables.dropped.forEach(this::dropTable);
@@ -685,10 +708,12 @@ public final class Schema
         delta.views.altered.forEach(diff -> notifyAlterView(diff.before, diff.after));
         delta.udfs.altered.forEach(diff -> notifyAlterFunction(diff.before, diff.after));
         delta.udas.altered.forEach(diff -> notifyAlterAggregate(diff.before, diff.after));
+        SchemaDiagnostics.keyspaceAltered(this, delta);
     }
 
     private void createKeyspace(KeyspaceMetadata keyspace)
     {
+        SchemaDiagnostics.keyspaceCreating(this, keyspace);
         load(keyspace);
         Keyspace.open(keyspace.name);
 
@@ -698,10 +723,12 @@ public final class Schema
         keyspace.views.forEach(this::notifyCreateView);
         keyspace.functions.udfs().forEach(this::notifyCreateFunction);
         keyspace.functions.udas().forEach(this::notifyCreateAggregate);
+        SchemaDiagnostics.keyspaceCreated(this, keyspace);
     }
 
     private void dropKeyspace(KeyspaceMetadata keyspace)
     {
+        SchemaDiagnostics.keyspaceDroping(this, keyspace);
         keyspace.views.forEach(this::dropView);
         keyspace.tables.forEach(this::dropTable);
 
@@ -716,6 +743,7 @@ public final class Schema
         keyspace.tables.forEach(this::notifyDropTable);
         keyspace.types.forEach(this::notifyDropType);
         notifyDropKeyspace(keyspace);
+        SchemaDiagnostics.keyspaceDroped(this, keyspace);
     }
 
     private void dropView(ViewMetadata metadata)
@@ -726,6 +754,7 @@ public final class Schema
 
     private void dropTable(TableMetadata metadata)
     {
+        SchemaDiagnostics.tableDropping(this, metadata);
         ColumnFamilyStore cfs = Keyspace.open(metadata.keyspace).getColumnFamilyStore(metadata.name);
         assert cfs != null;
         // make sure all the indexes are dropped, or else.
@@ -735,11 +764,14 @@ public final class Schema
             cfs.snapshot(Keyspace.getTimestampedSnapshotNameWithPrefix(cfs.name, ColumnFamilyStore.SNAPSHOT_DROP_PREFIX));
         CommitLog.instance.forceRecycleAllSegments(Collections.singleton(metadata.id));
         Keyspace.open(metadata.keyspace).dropCf(metadata.id);
+        SchemaDiagnostics.tableDropped(this, metadata);
     }
 
     private void createTable(TableMetadata table)
     {
+        SchemaDiagnostics.tableCreating(this, table);
         Keyspace.open(table.keyspace).initCf(metadataRefs.get(table.id), true);
+        SchemaDiagnostics.tableCreated(this, table);
     }
 
     private void createView(ViewMetadata view)
@@ -749,7 +781,9 @@ public final class Schema
 
     private void alterTable(TableMetadata updated)
     {
+        SchemaDiagnostics.tableAltering(this, updated);
         Keyspace.open(updated.keyspace).getColumnFamilyStore(updated.name).reload();
+        SchemaDiagnostics.tableAltered(this, updated);
     }
 
     private void alterView(ViewMetadata updated)

src/java/org/apache/cassandra/schema/SchemaAnnouncementDiagnostics.java

@@ -0,0 +1,60 @@
+/*
+ * 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.schema;
+
+import java.util.Set;
+
+import org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.schema.SchemaAnnouncementEvent.SchemaAnnouncementEventType;
+
+final class SchemaAnnouncementDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private SchemaAnnouncementDiagnostics()
+    {
+    }
+
+    static void schemaMutationsAnnounced(Set<InetAddressAndPort> schemaDestinationEndpoints, Set<InetAddressAndPort> schemaEndpointsIgnored)
+    {
+        if (isEnabled(SchemaAnnouncementEventType.SCHEMA_MUTATIONS_ANNOUNCED))
+            service.publish(new SchemaAnnouncementEvent(SchemaAnnouncementEventType.SCHEMA_MUTATIONS_ANNOUNCED,
+                                                        schemaDestinationEndpoints, schemaEndpointsIgnored, null, null));
+    }
+
+    public static void schemataMutationsReceived(InetAddressAndPort from)
+    {
+        if (isEnabled(SchemaAnnouncementEventType.SCHEMA_MUTATIONS_RECEIVED))
+            service.publish(new SchemaAnnouncementEvent(SchemaAnnouncementEventType.SCHEMA_MUTATIONS_RECEIVED,
+                                                        null, null, null, from));
+    }
+
+    static void schemaTransformationAnnounced(Set<InetAddressAndPort> schemaDestinationEndpoints, Set<InetAddressAndPort> schemaEndpointsIgnored, SchemaTransformation transformation)
+    {
+        if (isEnabled(SchemaAnnouncementEventType.SCHEMA_TRANSFORMATION_ANNOUNCED))
+            service.publish(new SchemaAnnouncementEvent(SchemaAnnouncementEventType.SCHEMA_TRANSFORMATION_ANNOUNCED,
+                                                        schemaDestinationEndpoints, schemaEndpointsIgnored, transformation, null));
+    }
+
+    private static boolean isEnabled(SchemaAnnouncementEventType type)
+    {
+        return service.isEnabled(SchemaAnnouncementEvent.class, type);
+    }
+}

src/java/org/apache/cassandra/schema/SchemaAnnouncementEvent.java

@@ -0,0 +1,104 @@
+/*
+ * 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.schema;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.audit.AuditLogContext;
+import org.apache.cassandra.cql3.CQLStatement;
+import org.apache.cassandra.diag.DiagnosticEvent;
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+/**
+ * Events emitted by {@link MigrationManager} around propagating schema changes to remote nodes.
+ */
+final class SchemaAnnouncementEvent extends DiagnosticEvent
+{
+    private final SchemaAnnouncementEventType type;
+    @Nullable
+    private final Set<InetAddressAndPort> schemaDestinationEndpoints;
+    @Nullable
+    private final Set<InetAddressAndPort> schemaEndpointsIgnored;
+    @Nullable
+    private final CQLStatement statement;
+    @Nullable
+    private final InetAddressAndPort sender;
+
+    enum SchemaAnnouncementEventType
+    {
+        SCHEMA_MUTATIONS_ANNOUNCED,
+        SCHEMA_TRANSFORMATION_ANNOUNCED,
+        SCHEMA_MUTATIONS_RECEIVED
+    }
+
+    SchemaAnnouncementEvent(SchemaAnnouncementEventType type,
+                            @Nullable Set<InetAddressAndPort> schemaDestinationEndpoints,
+                            @Nullable Set<InetAddressAndPort> schemaEndpointsIgnored,
+                            @Nullable SchemaTransformation transformation,
+                            @Nullable InetAddressAndPort sender)
+    {
+        this.type = type;
+        this.schemaDestinationEndpoints = schemaDestinationEndpoints;
+        this.schemaEndpointsIgnored = schemaEndpointsIgnored;
+        if (transformation instanceof CQLStatement) this.statement = (CQLStatement) transformation;
+        else this.statement = null;
+        this.sender = sender;
+    }
+
+    public Enum<?> getType()
+    {
+        return type;
+    }
+
+    public Map<String, Serializable> toMap()
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (schemaDestinationEndpoints != null)
+        {
+            Set<String> eps = schemaDestinationEndpoints.stream().map(InetAddressAndPort::toString).collect(Collectors.toSet());
+            ret.put("endpointDestinations", new HashSet<>(eps));
+        }
+        if (schemaEndpointsIgnored != null)
+        {
+            Set<String> eps = schemaEndpointsIgnored.stream().map(InetAddressAndPort::toString).collect(Collectors.toSet());
+            ret.put("endpointIgnored", new HashSet<>(eps));
+        }
+        if (statement != null)
+        {
+            AuditLogContext logContext = statement.getAuditLogContext();
+            if (logContext != null)
+            {
+                HashMap<String, String> log = new HashMap<>();
+                if (logContext.auditLogEntryType != null) log.put("type", logContext.auditLogEntryType.name());
+                if (logContext.keyspace != null) log.put("keyspace", logContext.keyspace);
+                if (logContext.scope != null) log.put("table", logContext.scope);
+                ret.put("statement", log);
+            }
+        }
+        if (sender != null) ret.put("sender", sender.toString());
+        return ret;
+    }
+}

src/java/org/apache/cassandra/schema/SchemaDiagnostics.java

@@ -0,0 +1,178 @@
+/*
+ * 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.schema;
+
+import com.google.common.collect.MapDifference;
+
+import org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.schema.SchemaEvent.SchemaEventType;
+
+final class SchemaDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private SchemaDiagnostics()
+    {
+    }
+
+    static void metadataInitialized(Schema schema, KeyspaceMetadata ksmUpdate)
+    {
+        if (isEnabled(SchemaEventType.KS_METADATA_LOADED))
+            service.publish(new SchemaEvent(SchemaEventType.KS_METADATA_LOADED, schema, ksmUpdate, null, null, null, null, null, null));
+    }
+
+    static void metadataReloaded(Schema schema, KeyspaceMetadata previous, KeyspaceMetadata ksmUpdate, Tables.TablesDiff tablesDiff, Views.ViewsDiff viewsDiff, MapDifference<String,TableMetadata> indexesDiff)
+    {
+        if (isEnabled(SchemaEventType.KS_METADATA_RELOADED))
+            service.publish(new SchemaEvent(SchemaEventType.KS_METADATA_RELOADED, schema, ksmUpdate, previous,
+                                            null, null, tablesDiff, viewsDiff, indexesDiff));
+    }
+
+    static void metadataRemoved(Schema schema, KeyspaceMetadata ksmUpdate)
+    {
+        if (isEnabled(SchemaEventType.KS_METADATA_REMOVED))
+            service.publish(new SchemaEvent(SchemaEventType.KS_METADATA_REMOVED, schema, ksmUpdate,
+                                            null, null, null, null, null, null));
+    }
+
+    static void versionUpdated(Schema schema)
+    {
+        if (isEnabled(SchemaEventType.VERSION_UPDATED))
+            service.publish(new SchemaEvent(SchemaEventType.VERSION_UPDATED, schema,
+                                            null, null, null, null, null, null, null));
+    }
+
+    static void keyspaceCreating(Schema schema, KeyspaceMetadata keyspace)
+    {
+        if (isEnabled(SchemaEventType.KS_CREATING))
+            service.publish(new SchemaEvent(SchemaEventType.KS_CREATING, schema, keyspace,
+                                            null, null, null, null, null, null));
+    }
+
+    static void keyspaceCreated(Schema schema, KeyspaceMetadata keyspace)
+    {
+        if (isEnabled(SchemaEventType.KS_CREATED))
+            service.publish(new SchemaEvent(SchemaEventType.KS_CREATED, schema, keyspace,
+                                            null, null, null, null, null, null));
+    }
+
+    static void keyspaceAltering(Schema schema, KeyspaceMetadata.KeyspaceDiff delta)
+    {
+        if (isEnabled(SchemaEventType.KS_ALTERING))
+            service.publish(new SchemaEvent(SchemaEventType.KS_ALTERING, schema, delta.after,
+                                            delta.before, delta, null, null, null, null));
+    }
+
+    static void keyspaceAltered(Schema schema, KeyspaceMetadata.KeyspaceDiff delta)
+    {
+        if (isEnabled(SchemaEventType.KS_ALTERED))
+            service.publish(new SchemaEvent(SchemaEventType.KS_ALTERED, schema, delta.after,
+                                            delta.before, delta, null, null, null, null));
+    }
+
+    static void keyspaceDroping(Schema schema, KeyspaceMetadata keyspace)
+    {
+        if (isEnabled(SchemaEventType.KS_DROPPING))
+            service.publish(new SchemaEvent(SchemaEventType.KS_DROPPING, schema, keyspace,
+                                            null, null, null, null, null, null));
+    }
+
+    static void keyspaceDroped(Schema schema, KeyspaceMetadata keyspace)
+    {
+        if (isEnabled(SchemaEventType.KS_DROPPED))
+            service.publish(new SchemaEvent(SchemaEventType.KS_DROPPED, schema, keyspace,
+                                            null, null, null, null, null, null));
+    }
+
+    static void schemataLoading(Schema schema)
+    {
+        if (isEnabled(SchemaEventType.SCHEMATA_LOADING))
+            service.publish(new SchemaEvent(SchemaEventType.SCHEMATA_LOADING, schema, null,
+                                            null, null, null, null, null, null));
+    }
+
+    static void schemataLoaded(Schema schema)
+    {
+        if (isEnabled(SchemaEventType.SCHEMATA_LOADED))
+            service.publish(new SchemaEvent(SchemaEventType.SCHEMATA_LOADED, schema, null,
+                                            null, null, null, null, null, null));
+    }
+
+    static void versionAnnounced(Schema schema)
+    {
+        if (isEnabled(SchemaEventType.VERSION_ANOUNCED))
+            service.publish(new SchemaEvent(SchemaEventType.VERSION_ANOUNCED, schema, null,
+                                            null, null, null, null, null, null));
+    }
+
+    static void schemataCleared(Schema schema)
+    {
+        if (isEnabled(SchemaEventType.SCHEMATA_CLEARED))
+            service.publish(new SchemaEvent(SchemaEventType.SCHEMATA_CLEARED, schema, null,
+                                            null, null, null, null, null, null));
+    }
+
+    static void tableCreating(Schema schema, TableMetadata table)
+    {
+        if (isEnabled(SchemaEventType.TABLE_CREATING))
+            service.publish(new SchemaEvent(SchemaEventType.TABLE_CREATING, schema, null,
+                                            null, null, table, null, null, null));
+    }
+
+    static void tableCreated(Schema schema, TableMetadata table)
+    {
+        if (isEnabled(SchemaEventType.TABLE_CREATED))
+            service.publish(new SchemaEvent(SchemaEventType.TABLE_CREATED, schema, null,
+                                            null, null, table, null, null, null));
+    }
+
+    static void tableAltering(Schema schema, TableMetadata table)
+    {
+        if (isEnabled(SchemaEventType.TABLE_ALTERING))
+            service.publish(new SchemaEvent(SchemaEventType.TABLE_ALTERING, schema, null,
+                                            null, null, table, null, null, null));
+    }
+
+    static void tableAltered(Schema schema, TableMetadata table)
+    {
+        if (isEnabled(SchemaEventType.TABLE_ALTERED))
+            service.publish(new SchemaEvent(SchemaEventType.TABLE_ALTERED, schema, null,
+                                            null, null, table, null, null, null));
+    }
+
+    static void tableDropping(Schema schema, TableMetadata table)
+    {
+        if (isEnabled(SchemaEventType.TABLE_DROPPING))
+            service.publish(new SchemaEvent(SchemaEventType.TABLE_DROPPING, schema, null,
+                                            null, null, table, null, null, null));
+    }
+
+    static void tableDropped(Schema schema, TableMetadata table)
+    {
+        if (isEnabled(SchemaEventType.TABLE_DROPPED))
+            service.publish(new SchemaEvent(SchemaEventType.TABLE_DROPPED, schema, null,
+                                            null, null, table, null, null, null));
+    }
+
+    private static boolean isEnabled(SchemaEventType type)
+    {
+        return service.isEnabled(SchemaEvent.class, type);
+    }
+
+}

src/java/org/apache/cassandra/schema/SchemaEvent.java

@@ -0,0 +1,318 @@
+/*
+ * 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.schema;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.MapDifference;
+
+import org.apache.cassandra.diag.DiagnosticEvent;
+import org.apache.cassandra.utils.Pair;
+
+final class SchemaEvent extends DiagnosticEvent
+{
+    private final SchemaEventType type;
+
+    private final HashSet<String> keyspaces;
+    private final HashMap<String, String> indexTables;
+    private final HashMap<String, String> tables;
+    private final ArrayList<String> nonSystemKeyspaces;
+    private final ArrayList<String> userKeyspaces;
+    private final int numberOfTables;
+    private final UUID version;
+
+    @Nullable
+    private final KeyspaceMetadata ksUpdate;
+    @Nullable
+    private final KeyspaceMetadata previous;
+    @Nullable
+    private final KeyspaceMetadata.KeyspaceDiff ksDiff;
+    @Nullable
+    private final TableMetadata tableUpdate;
+    @Nullable
+    private final Tables.TablesDiff tablesDiff;
+    @Nullable
+    private final Views.ViewsDiff viewsDiff;
+    @Nullable
+    private final MapDifference<String,TableMetadata> indexesDiff;
+
+    enum SchemaEventType
+    {
+        KS_METADATA_LOADED,
+        KS_METADATA_RELOADED,
+        KS_METADATA_REMOVED,
+        VERSION_UPDATED,
+        VERSION_ANOUNCED,
+        KS_CREATING,
+        KS_CREATED,
+        KS_ALTERING,
+        KS_ALTERED,
+        KS_DROPPING,
+        KS_DROPPED,
+        TABLE_CREATING,
+        TABLE_CREATED,
+        TABLE_ALTERING,
+        TABLE_ALTERED,
+        TABLE_DROPPING,
+        TABLE_DROPPED,
+        SCHEMATA_LOADING,
+        SCHEMATA_LOADED,
+        SCHEMATA_CLEARED
+    }
+
+    SchemaEvent(SchemaEventType type, Schema schema, @Nullable KeyspaceMetadata ksUpdate,
+                @Nullable KeyspaceMetadata previous, @Nullable KeyspaceMetadata.KeyspaceDiff ksDiff,
+                @Nullable TableMetadata tableUpdate, @Nullable Tables.TablesDiff tablesDiff,
+                @Nullable Views.ViewsDiff viewsDiff, @Nullable MapDifference<String,TableMetadata> indexesDiff)
+    {
+        this.type = type;
+        this.ksUpdate = ksUpdate;
+        this.previous = previous;
+        this.ksDiff = ksDiff;
+        this.tableUpdate = tableUpdate;
+        this.tablesDiff = tablesDiff;
+        this.viewsDiff = viewsDiff;
+        this.indexesDiff = indexesDiff;
+
+        this.keyspaces = new HashSet<>(schema.getKeyspaces());
+        this.nonSystemKeyspaces = new ArrayList<>(schema.getNonSystemKeyspaces());
+        this.userKeyspaces = new ArrayList<>(schema.getUserKeyspaces());
+        this.numberOfTables = schema.getNumberOfTables();
+        this.version = schema.getVersion();
+
+        Map<Pair<String, String>, TableMetadataRef> indexTableMetadataRefs = schema.getIndexTableMetadataRefs();
+        Map<String, String> indexTables = indexTableMetadataRefs.entrySet().stream()
+                                                                .collect(Collectors.toMap(e -> e.getKey().left + ',' +
+                                                                                               e.getKey().right,
+                                                                                          e -> e.getValue().id.toHexString() + ',' +
+                                                                                               e.getValue().keyspace + ',' +
+                                                                                               e.getValue().name));
+        this.indexTables = new HashMap<>(indexTables);
+        Map<TableId, TableMetadataRef> tableMetadataRefs = schema.getTableMetadataRefs();
+        Map<String, String> tables = tableMetadataRefs.entrySet().stream()
+                                                      .collect(Collectors.toMap(e -> e.getKey().toHexString(),
+                                                                                e -> e.getValue().id.toHexString() + ',' +
+                                                                                     e.getValue().keyspace + ',' +
+                                                                                     e.getValue().name));
+        this.tables = new HashMap<>(tables);
+    }
+
+    public SchemaEventType getType()
+    {
+        return type;
+    }
+
+    public Map<String, Serializable> toMap()
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        ret.put("keyspaces", this.keyspaces);
+        ret.put("nonSystemKeyspaces", this.nonSystemKeyspaces);
+        ret.put("userKeyspaces", this.userKeyspaces);
+        ret.put("numberOfTables", this.numberOfTables);
+        ret.put("version", this.version);
+        ret.put("tables", this.tables);
+        ret.put("indexTables", this.indexTables);
+        if (ksUpdate != null) ret.put("ksMetadataUpdate", repr(ksUpdate));
+        if (previous != null) ret.put("ksMetadataPrevious", repr(previous));
+        if (ksDiff != null)
+        {
+            HashMap<String, Serializable> ks = new HashMap<>();
+            ks.put("before", repr(ksDiff.before));
+            ks.put("after", repr(ksDiff.after));
+            ks.put("tables", repr(ksDiff.tables));
+            ks.put("views", repr(ksDiff.views));
+            ks.put("types", repr(ksDiff.types));
+            ks.put("udas", repr(ksDiff.udas));
+            ks.put("udfs", repr(ksDiff.udfs));
+            ret.put("ksDiff", ks);
+        }
+        if (tableUpdate != null) ret.put("tableMetadataUpdate", repr(tableUpdate));
+        if (tablesDiff != null) ret.put("tablesDiff", repr(tablesDiff));
+        if (viewsDiff != null) ret.put("viewsDiff", repr(viewsDiff));
+        if (indexesDiff != null) ret.put("indexesDiff", Lists.newArrayList(indexesDiff.entriesDiffering().keySet()));
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(Diff<?, ?> diff)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (diff.created != null) ret.put("created", diff.created.toString());
+        if (diff.dropped != null) ret.put("dropped", diff.dropped.toString());
+        if (diff.altered != null)
+            ret.put("created", Lists.newArrayList(diff.altered.stream().map(Diff.Altered::toString).iterator()));
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(KeyspaceMetadata ksm)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        ret.put("name", ksm.name);
+        if (ksm.kind != null) ret.put("kind", ksm.kind.name());
+        if (ksm.params != null) ret.put("params", ksm.params.toString());
+        if (ksm.tables != null) ret.put("tables", ksm.tables.toString());
+        if (ksm.views != null) ret.put("views", ksm.views.toString());
+        if (ksm.functions != null) ret.put("functions", ksm.functions.toString());
+        if (ksm.types != null) ret.put("types", ksm.types.toString());
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(TableMetadata table)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        ret.put("id", table.id.toHexString());
+        ret.put("name", table.name);
+        ret.put("keyspace", table.keyspace);
+        ret.put("partitioner", table.partitioner.toString());
+        ret.put("kind", table.kind.name());
+        ret.put("flags", Lists.newArrayList(table.flags.stream().map(Enum::name).iterator()));
+        ret.put("params", repr(table.params));
+        ret.put("indexes", Lists.newArrayList(table.indexes.stream().map(this::repr).iterator()));
+        ret.put("triggers", Lists.newArrayList(repr(table.triggers)));
+        ret.put("columns", Lists.newArrayList(table.columns.values().stream().map(this::repr).iterator()));
+        ret.put("droppedColumns", Lists.newArrayList(table.droppedColumns.values().stream().map(this::repr).iterator()));
+        ret.put("isCompactTable", table.isCompactTable());
+        ret.put("isCompound", table.isCompound());
+        ret.put("isCounter", table.isCounter());
+        ret.put("isCQLTable", table.isCQLTable());
+        ret.put("isDense", table.isDense());
+        ret.put("isIndex", table.isIndex());
+        ret.put("isStaticCompactTable", table.isStaticCompactTable());
+        ret.put("isSuper", table.isSuper());
+        ret.put("isView", table.isView());
+        ret.put("isVirtual", table.isVirtual());
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(TableParams params)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (params == null) return ret;
+        ret.put("minIndexInterval", params.minIndexInterval);
+        ret.put("maxIndexInterval", params.maxIndexInterval);
+        ret.put("defaultTimeToLive", params.defaultTimeToLive);
+        ret.put("gcGraceSeconds", params.gcGraceSeconds);
+        ret.put("bloomFilterFpChance", params.bloomFilterFpChance);
+        ret.put("cdc", params.cdc);
+        ret.put("crcCheckChance", params.crcCheckChance);
+        ret.put("memtableFlushPeriodInMs", params.memtableFlushPeriodInMs);
+        ret.put("comment", params.comment);
+        ret.put("caching", repr(params.caching));
+        ret.put("compaction", repr(params.compaction));
+        ret.put("compression", repr(params.compression));
+        if (params.speculativeRetry != null) ret.put("speculativeRetry", params.speculativeRetry.kind().name());
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(CachingParams caching)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (caching == null) return ret;
+        ret.putAll(caching.asMap());
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(CompactionParams comp)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (comp == null) return ret;
+        ret.putAll(comp.asMap());
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(CompressionParams compr)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (compr == null) return ret;
+        ret.putAll(compr.asMap());
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(IndexMetadata index)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (index == null) return ret;
+        ret.put("name", index.name);
+        ret.put("kind", index.kind.name());
+        ret.put("id", index.id);
+        ret.put("options", new HashMap<>(index.options));
+        ret.put("isCustom", index.isCustom());
+        ret.put("isKeys", index.isKeys());
+        ret.put("isComposites", index.isComposites());
+        return ret;
+    }
+
+    private List<Map<String, Serializable>> repr(Triggers triggers)
+    {
+        List<Map<String, Serializable>> ret = new ArrayList<>();
+        if (triggers == null) return ret;
+        Iterator<TriggerMetadata> iter = triggers.iterator();
+        while (iter.hasNext()) ret.add(repr(iter.next()));
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(TriggerMetadata trigger)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (trigger == null) return ret;
+        ret.put("name", trigger.name);
+        ret.put("classOption", trigger.classOption);
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(ColumnMetadata col)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (col == null) return ret;
+        ret.put("name", col.name.toString());
+        ret.put("kind", col.kind.name());
+        ret.put("type", col.type.toString());
+        ret.put("ksName", col.ksName);
+        ret.put("cfName", col.cfName);
+        ret.put("position", col.position());
+        ret.put("clusteringOrder", col.clusteringOrder().name());
+        ret.put("isComplex", col.isComplex());
+        ret.put("isStatic", col.isStatic());
+        ret.put("isPrimaryKeyColumn", col.isPrimaryKeyColumn());
+        ret.put("isSimple", col.isSimple());
+        ret.put("isPartitionKey", col.isPartitionKey());
+        ret.put("isClusteringColumn", col.isClusteringColumn());
+        ret.put("isCounterColumn", col.isCounterColumn());
+        ret.put("isRegular", col.isRegular());
+        return ret;
+    }
+
+    private HashMap<String, Serializable> repr(DroppedColumn column)
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (column == null) return ret;
+        ret.put("droppedTime", column.droppedTime);
+        ret.put("column", repr(column.column));
+        return ret;
+    }
+}

src/java/org/apache/cassandra/schema/SchemaMigrationDiagnostics.java

@@ -0,0 +1,83 @@
+/*
+ * 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.schema;
+
+import java.util.UUID;
+
+import org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.schema.SchemaMigrationEvent.MigrationManagerEventType;
+
+final class SchemaMigrationDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private SchemaMigrationDiagnostics()
+    {
+    }
+
+    static void unknownLocalSchemaVersion(InetAddressAndPort endpoint, UUID theirVersion)
+    {
+        if (isEnabled(MigrationManagerEventType.UNKNOWN_LOCAL_SCHEMA_VERSION))
+            service.publish(new SchemaMigrationEvent(MigrationManagerEventType.UNKNOWN_LOCAL_SCHEMA_VERSION, endpoint,
+                                                     theirVersion));
+    }
+
+    static void versionMatch(InetAddressAndPort endpoint, UUID theirVersion)
+    {
+        if (isEnabled(MigrationManagerEventType.VERSION_MATCH))
+            service.publish(new SchemaMigrationEvent(MigrationManagerEventType.VERSION_MATCH, endpoint, theirVersion));
+    }
+
+    static void skipPull(InetAddressAndPort endpoint, UUID theirVersion)
+    {
+        if (isEnabled(MigrationManagerEventType.SKIP_PULL))
+            service.publish(new SchemaMigrationEvent(MigrationManagerEventType.SKIP_PULL, endpoint, theirVersion));
+    }
+
+    static void resetLocalSchema()
+    {
+        if (isEnabled(MigrationManagerEventType.RESET_LOCAL_SCHEMA))
+            service.publish(new SchemaMigrationEvent(MigrationManagerEventType.RESET_LOCAL_SCHEMA, null, null));
+    }
+
+    static void taskCreated(InetAddressAndPort endpoint)
+    {
+        if (isEnabled(MigrationManagerEventType.TASK_CREATED))
+            service.publish(new SchemaMigrationEvent(MigrationManagerEventType.TASK_CREATED, endpoint, null));
+    }
+
+    static void taskSendAborted(InetAddressAndPort endpoint)
+    {
+        if (isEnabled(MigrationManagerEventType.TASK_SEND_ABORTED))
+            service.publish(new SchemaMigrationEvent(MigrationManagerEventType.TASK_SEND_ABORTED, endpoint, null));
+    }
+
+    static void taskRequestSend(InetAddressAndPort endpoint)
+    {
+        if (isEnabled(MigrationManagerEventType.TASK_REQUEST_SEND))
+            service.publish(new SchemaMigrationEvent(MigrationManagerEventType.TASK_REQUEST_SEND,
+                                                     endpoint, null));
+    }
+
+    private static boolean isEnabled(MigrationManagerEventType type)
+    {
+        return service.isEnabled(SchemaMigrationEvent.class, type);
+    }
+}

src/java/org/apache/cassandra/schema/SchemaMigrationEvent.java

@@ -0,0 +1,114 @@
+/*
+ * 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.schema;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.diag.DiagnosticEvent;
+import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.MessagingService;
+
+/**
+ * Internal events emitted by {@link MigrationManager}.
+ */
+final class SchemaMigrationEvent extends DiagnosticEvent
+{
+    private final MigrationManagerEventType type;
+    @Nullable
+    private final InetAddressAndPort endpoint;
+    @Nullable
+    private final UUID endpointSchemaVersion;
+    private final UUID localSchemaVersion;
+    private final Integer localMessagingVersion;
+    private final SystemKeyspace.BootstrapState bootstrapState;
+    @Nullable
+    private Integer inflightTaskCount;
+    @Nullable
+    private Integer endpointMessagingVersion;
+    @Nullable
+    private Boolean endpointGossipOnlyMember;
+    @Nullable
+    private Boolean isAlive;
+
+    enum MigrationManagerEventType
+    {
+        UNKNOWN_LOCAL_SCHEMA_VERSION,
+        VERSION_MATCH,
+        SKIP_PULL,
+        RESET_LOCAL_SCHEMA,
+        TASK_CREATED,
+        TASK_SEND_ABORTED,
+        TASK_REQUEST_SEND
+    }
+
+    SchemaMigrationEvent(MigrationManagerEventType type,
+                         @Nullable InetAddressAndPort endpoint, @Nullable UUID endpointSchemaVersion)
+    {
+        this.type = type;
+        this.endpoint = endpoint;
+        this.endpointSchemaVersion = endpointSchemaVersion;
+
+        localSchemaVersion = Schema.instance.getVersion();
+        localMessagingVersion = MessagingService.current_version;
+
+        Queue<CountDownLatch> inflightTasks = MigrationTask.getInflightTasks();
+        if (inflightTasks != null)
+            inflightTaskCount = inflightTasks.size();
+
+        this.bootstrapState = SystemKeyspace.getBootstrapState();
+
+        if (endpoint == null) return;
+
+        if (MessagingService.instance().knowsVersion(endpoint))
+            endpointMessagingVersion = MessagingService.instance().getRawVersion(endpoint);
+
+        endpointGossipOnlyMember = Gossiper.instance.isGossipOnlyMember(endpoint);
+        this.isAlive = FailureDetector.instance.isAlive(endpoint);
+    }
+
+    public Enum<?> getType()
+    {
+        return type;
+    }
+
+    public Map<String, Serializable> toMap()
+    {
+        HashMap<String, Serializable> ret = new HashMap<>();
+        if (endpoint != null) ret.put("endpoint", endpoint.getHostAddress(true));
+        ret.put("endpointSchemaVersion", Schema.schemaVersionToString(endpointSchemaVersion));
+        ret.put("localSchemaVersion", Schema.schemaVersionToString(localSchemaVersion));
+        if (endpointMessagingVersion != null) ret.put("endpointMessagingVersion", endpointMessagingVersion);
+        if (localMessagingVersion != null) ret.put("localMessagingVersion", localMessagingVersion);
+        if (endpointGossipOnlyMember != null) ret.put("endpointGossipOnlyMember", endpointGossipOnlyMember);
+        if (isAlive != null) ret.put("endpointIsAlive", isAlive);
+        if (bootstrapState != null) ret.put("bootstrapState", bootstrapState.name());
+        if (inflightTaskCount != null) ret.put("inflightTaskCount", inflightTaskCount);
+        return ret;
+    }
+}

src/java/org/apache/cassandra/schema/SchemaPushVerbHandler.java

@@ -42,6 +42,7 @@ public final class SchemaPushVerbHandler implements IVerbHandler<Collection<Muta
     {
         logger.trace("Received schema push request from {}", message.from);
 
+        SchemaAnnouncementDiagnostics.schemataMutationsReceived(message.from);
         StageManager.getStage(Stage.MIGRATION).submit(() -> Schema.instance.mergeAndAnnounceVersion(message.payload));
     }
 }

src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java

@@ -45,28 +45,35 @@ public class PendingRangeCalculatorService
 
     public PendingRangeCalculatorService()
     {
-        executor.setRejectedExecutionHandler(new RejectedExecutionHandler()
-        {
-            public void rejectedExecution(Runnable r, ThreadPoolExecutor e)
+        executor.setRejectedExecutionHandler((r, e) ->
             {
+                PendingRangeCalculatorServiceDiagnostics.taskRejected(instance, updateJobs);
                 PendingRangeCalculatorService.instance.finishUpdate();
             }
-        }
         );
     }
 
     private static class PendingRangeTask implements Runnable
     {
+        private final AtomicInteger updateJobs;
+
+        PendingRangeTask(AtomicInteger updateJobs)
+        {
+            this.updateJobs = updateJobs;
+        }
+
         public void run()
         {
             try
             {
+                PendingRangeCalculatorServiceDiagnostics.taskStarted(instance, updateJobs);
                 long start = System.currentTimeMillis();
                 List<String> keyspaces = Schema.instance.getNonLocalStrategyKeyspaces();
                 for (String keyspaceName : keyspaces)
                     calculatePendingRanges(Keyspace.open(keyspaceName).getReplicationStrategy(), keyspaceName);
                 if (logger.isTraceEnabled())
                     logger.trace("Finished PendingRangeTask for {} keyspaces in {}ms", keyspaces.size(), System.currentTimeMillis() - start);
+                PendingRangeCalculatorServiceDiagnostics.taskFinished(instance, updateJobs);
             }
             finally
             {
@@ -77,13 +84,15 @@ public class PendingRangeCalculatorService
 
     private void finishUpdate()
     {
-        updateJobs.decrementAndGet();
+        int jobs = updateJobs.decrementAndGet();
+        PendingRangeCalculatorServiceDiagnostics.taskCountChanged(instance, jobs);
     }
 
     public void update()
     {
-        updateJobs.incrementAndGet();
-        executor.submit(new PendingRangeTask());
+        int jobs = updateJobs.incrementAndGet();
+        PendingRangeCalculatorServiceDiagnostics.taskCountChanged(instance, jobs);
+        executor.submit(new PendingRangeTask(updateJobs));
     }
 
     public void blockUntilFinished()

src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceDiagnostics.java

@@ -0,0 +1,73 @@
+/*
+ * 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.service;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.cassandra.diag.DiagnosticEventService;
+import org.apache.cassandra.service.PendingRangeCalculatorServiceEvent.PendingRangeCalculatorServiceEventType;
+
+/**
+ * Utility methods for diagnostic events related to {@link PendingRangeCalculatorService}.
+ */
+final class PendingRangeCalculatorServiceDiagnostics
+{
+    private static final DiagnosticEventService service = DiagnosticEventService.instance();
+
+    private PendingRangeCalculatorServiceDiagnostics()
+    {
+    }
+    
+    static void taskStarted(PendingRangeCalculatorService calculatorService, AtomicInteger taskCount)
+    {
+        if (isEnabled(PendingRangeCalculatorServiceEventType.TASK_STARTED))
+            service.publish(new PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType.TASK_STARTED,
+                                                                   calculatorService,
+                                                                   taskCount.get()));
+    }
+
+    static void taskFinished(PendingRangeCalculatorService calculatorService, AtomicInteger taskCount)
+    {
+        if (isEnabled(PendingRangeCalculatorServiceEventType.TASK_FINISHED_SUCCESSFULLY))
+            service.publish(new PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType.TASK_FINISHED_SUCCESSFULLY,
+                                                                   calculatorService,
+                                                                   taskCount.get()));
+    }
+
+    static void taskRejected(PendingRangeCalculatorService calculatorService, AtomicInteger taskCount)
+    {
+        if (isEnabled(PendingRangeCalculatorServiceEventType.TASK_EXECUTION_REJECTED))
+            service.publish(new PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType.TASK_EXECUTION_REJECTED,
+                                                                   calculatorService,
+                                                                   taskCount.get()));
+    }
+
+    static void taskCountChanged(PendingRangeCalculatorService calculatorService, int taskCount)
+    {
+        if (isEnabled(PendingRangeCalculatorServiceEventType.TASK_COUNT_CHANGED))
+            service.publish(new PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType.TASK_COUNT_CHANGED,
+                                                                   calculatorService,
+                                                                   taskCount));
+    }
+
+    private static boolean isEnabled(PendingRangeCalculatorServiceEventType type)
+    {
+        return service.isEnabled(PendingRangeCalculatorServiceEvent.class, type);
+    }
+}

src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceEvent.java

@@ -0,0 +1,69 @@
+/*
+ * 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.service;
+
+import java.io.Serializable;
+import java.util.HashMap;
+
+import org.apache.cassandra.diag.DiagnosticEvent;
+
+/**
+ * Events related to {@link PendingRangeCalculatorService}.
+ */
+final class PendingRangeCalculatorServiceEvent extends DiagnosticEvent
+{
+    private final PendingRangeCalculatorServiceEventType type;
+    private final PendingRangeCalculatorService source;
+    private final int taskCount;
+
+    public enum PendingRangeCalculatorServiceEventType
+    {
+        TASK_STARTED,
+        TASK_FINISHED_SUCCESSFULLY,
+        TASK_EXECUTION_REJECTED,
+        TASK_COUNT_CHANGED
+    }
+
+    PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType type,
+                                       PendingRangeCalculatorService service,
+                                       int taskCount)
+    {
+        this.type = type;
+        this.source = service;
+        this.taskCount = taskCount;
+    }
+
+    public int getTaskCount()
+    {
+        return taskCount;
+    }
+
+    public PendingRangeCalculatorServiceEventType getType()
+    {
+        return type;
+    }
+
+    public HashMap<String, Serializable> toMap()
+    {
+        // be extra defensive against nulls and bugs
+        HashMap<String, Serializable> ret = new HashMap<>();
+        ret.put("taskCount", taskCount);
+        return ret;
+    }
+}

test/unit/org/apache/cassandra/diag/DiagnosticEventServiceTest.java

@@ -0,0 +1,244 @@
+/*
+ * 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.diag;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.function.Consumer;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.After;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.apache.cassandra.OrderedJUnit4ClassRunner;
+import org.apache.cassandra.config.DatabaseDescriptor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(OrderedJUnit4ClassRunner.class)
+public class DiagnosticEventServiceTest
+{
+
+    @BeforeClass
+    public static void setup()
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @After
+    public void cleanup()
+    {
+        DiagnosticEventService.instance().cleanup();
+    }
+
+    @Test
+    public void testSubscribe()
+    {
+        DiagnosticEventService instance = DiagnosticEventService.instance();
+        assertFalse(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+        Consumer<TestEvent1> consumer1 = (event) ->
+        {
+        };
+        Consumer<TestEvent1> consumer2 = (event) ->
+        {
+        };
+        Consumer<TestEvent1> consumer3 = (event) ->
+        {
+        };
+        instance.subscribe(TestEvent1.class, consumer1);
+        instance.subscribe(TestEvent1.class, consumer2);
+        instance.subscribe(TestEvent1.class, consumer3);
+        assertTrue(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+        instance.unsubscribe(consumer1);
+        assertTrue(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+        instance.unsubscribe(consumer2);
+        instance.unsubscribe(consumer3);
+        assertFalse(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+    }
+
+    @Test
+    public void testSubscribeByType()
+    {
+        DiagnosticEventService instance = DiagnosticEventService.instance();
+        assertFalse(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+        Consumer<TestEvent1> consumer1 = (event) ->
+        {
+        };
+        Consumer<TestEvent1> consumer2 = (event) ->
+        {
+        };
+        Consumer<TestEvent1> consumer3 = (event) ->
+        {
+        };
+
+        assertFalse(instance.hasSubscribers(TestEvent1.class, TestEventType.TEST1));
+        instance.subscribe(TestEvent1.class, TestEventType.TEST1, consumer1);
+        assertTrue(instance.hasSubscribers(TestEvent1.class, TestEventType.TEST1));
+        assertFalse(instance.hasSubscribers(TestEvent1.class, TestEventType.TEST2));
+
+        instance.subscribe(TestEvent1.class, TestEventType.TEST2, consumer2);
+        instance.subscribe(TestEvent1.class, TestEventType.TEST2, consumer2);
+        instance.subscribe(TestEvent1.class, TestEventType.TEST2, consumer2);
+        assertTrue(instance.hasSubscribers(TestEvent1.class, TestEventType.TEST2));
+
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+
+        instance.subscribe(TestEvent1.class, consumer3);
+        assertTrue(instance.hasSubscribers(TestEvent1.class));
+        assertTrue(instance.hasSubscribers(TestEvent1.class, TestEventType.TEST1));
+        assertTrue(instance.hasSubscribers(TestEvent1.class, TestEventType.TEST2));
+        assertTrue(instance.hasSubscribers(TestEvent1.class, TestEventType.TEST3));
+
+        instance.unsubscribe(consumer1);
+        assertTrue(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+        instance.unsubscribe(consumer2);
+        instance.unsubscribe(consumer3);
+        assertFalse(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+    }
+
+    @Test
+    public void testSubscribeAll()
+    {
+        DiagnosticEventService instance = DiagnosticEventService.instance();
+        assertFalse(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+        Consumer<DiagnosticEvent> consumerAll1 = (event) ->
+        {
+        };
+        Consumer<DiagnosticEvent> consumerAll2 = (event) ->
+        {
+        };
+        Consumer<DiagnosticEvent> consumerAll3 = (event) ->
+        {
+        };
+        instance.subscribeAll(consumerAll1);
+        instance.subscribeAll(consumerAll2);
+        instance.subscribeAll(consumerAll3);
+        assertTrue(instance.hasSubscribers(TestEvent1.class));
+        assertTrue(instance.hasSubscribers(TestEvent2.class));
+        instance.unsubscribe(consumerAll1);
+        assertTrue(instance.hasSubscribers(TestEvent1.class));
+        assertTrue(instance.hasSubscribers(TestEvent2.class));
+        instance.unsubscribe(consumerAll2);
+        instance.unsubscribe(consumerAll3);
+        assertFalse(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+    }
+
+    @Test
+    public void testCleanup()
+    {
+        DiagnosticEventService instance = DiagnosticEventService.instance();
+        assertFalse(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+        Consumer<TestEvent1> consumer = (event) ->
+        {
+        };
+        instance.subscribe(TestEvent1.class, consumer);
+        Consumer<DiagnosticEvent> consumerAll = (event) ->
+        {
+        };
+        instance.subscribeAll(consumerAll);
+        assertTrue(instance.hasSubscribers(TestEvent1.class));
+        assertTrue(instance.hasSubscribers(TestEvent2.class));
+        instance.cleanup();
+        assertFalse(instance.hasSubscribers(TestEvent1.class));
+        assertFalse(instance.hasSubscribers(TestEvent2.class));
+    }
+
+    @Test
+    public void testPublish()
+    {
+        DiagnosticEventService instance = DiagnosticEventService.instance();
+        TestEvent1 a = new TestEvent1();
+        TestEvent1 b = new TestEvent1();
+        TestEvent1 c = new TestEvent1();
+        List<TestEvent1> events = ImmutableList.of(a, b, c, c, c);
+
+        List<DiagnosticEvent> consumed = new LinkedList<>();
+        Consumer<TestEvent1> consumer = consumed::add;
+        Consumer<DiagnosticEvent> consumerAll = consumed::add;
+
+        DatabaseDescriptor.setDiagnosticEventsEnabled(true);
+        instance.publish(c);
+        instance.subscribe(TestEvent1.class, consumer);
+        instance.publish(a);
+        instance.unsubscribe(consumer);
+        instance.publish(c);
+        instance.subscribeAll(consumerAll);
+        instance.publish(b);
+        instance.subscribe(TestEvent1.class, TestEventType.TEST3, consumer);
+        instance.publish(c);
+        instance.subscribe(TestEvent1.class, TestEventType.TEST1, consumer);
+        instance.publish(c);
+
+        assertEquals(events, consumed);
+    }
+
+    @Test
+    public void testEnabled()
+    {
+        DatabaseDescriptor.setDiagnosticEventsEnabled(false);
+        DiagnosticEventService.instance().subscribe(TestEvent1.class, (event) -> fail());
+        DiagnosticEventService.instance().publish(new TestEvent1());
+        DatabaseDescriptor.setDiagnosticEventsEnabled(true);
+    }
+
+    private static class TestEvent1 extends DiagnosticEvent
+    {
+        public TestEventType getType()
+        {
+            return TestEventType.TEST1;
+        }
+
+        public HashMap<String, Serializable> toMap()
+        {
+            return null;
+        }
+    }
+
+    private static class TestEvent2 extends DiagnosticEvent
+    {
+        public TestEventType getType()
+        {
+            return TestEventType.TEST2;
+        }
+
+        public HashMap<String, Serializable> toMap()
+        {
+            return null;
+        }
+    }
+
+    private enum TestEventType { TEST1, TEST2, TEST3 };
+}