From 2846b22a70d48bae25203be945e02dd3b6cfda56 Mon Sep 17 00:00:00 2001 From: Stefan Podkowinski Date: Thu, 16 Mar 2017 12:50:52 +0100 Subject: [PATCH] Add diagnostic events base classes patch by Stefan Podkowinski; reviewed by Mick Semb Wever for CASSANDRA-13457 --- CHANGES.txt | 1 + conf/cassandra.yaml | 5 + .../org/apache/cassandra/config/Config.java | 3 + .../cassandra/config/DatabaseDescriptor.java | 11 + .../schema/AlterKeyspaceStatement.java | 5 + .../schema/AlterTableStatement.java | 5 + .../statements/schema/AlterTypeStatement.java | 5 + .../statements/schema/AlterViewStatement.java | 5 + .../schema/CreateAggregateStatement.java | 5 + .../schema/CreateFunctionStatement.java | 5 + .../schema/CreateIndexStatement.java | 5 + .../schema/CreateKeyspaceStatement.java | 5 + .../schema/CreateTableStatement.java | 5 + .../schema/CreateTriggerStatement.java | 5 + .../schema/CreateTypeStatement.java | 5 + .../schema/CreateViewStatement.java | 5 + .../schema/DropAggregateStatement.java | 5 + .../schema/DropFunctionStatement.java | 5 + .../statements/schema/DropIndexStatement.java | 5 + .../schema/DropKeyspaceStatement.java | 5 + .../statements/schema/DropTableStatement.java | 5 + .../schema/DropTriggerStatement.java | 5 + .../statements/schema/DropTypeStatement.java | 5 + .../statements/schema/DropViewStatement.java | 5 + .../apache/cassandra/dht/BootStrapper.java | 14 +- .../cassandra/dht/BootstrapDiagnostics.java | 80 +++++ .../apache/cassandra/dht/BootstrapEvent.java | 82 +++++ .../NoReplicationTokenAllocator.java | 4 + .../ReplicationAwareTokenAllocator.java | 7 +- .../TokenAllocatorDiagnostics.java | 195 +++++++++++ .../tokenallocator/TokenAllocatorEvent.java | 113 +++++++ .../tokenallocator/TokenAllocatorFactory.java | 8 +- .../cassandra/diag/DiagnosticEvent.java | 50 +++ .../diag/DiagnosticEventService.java | 291 ++++++++++++++++ .../org/apache/cassandra/gms/Gossiper.java | 46 ++- .../cassandra/gms/GossiperDiagnostics.java | 113 +++++++ .../apache/cassandra/gms/GossiperEvent.java | 111 ++++++ src/java/org/apache/cassandra/hints/Hint.java | 2 +- .../cassandra/hints/HintDiagnostics.java | 85 +++++ .../org/apache/cassandra/hints/HintEvent.java | 102 ++++++ .../hints/HintsDispatchExecutor.java | 10 + .../cassandra/hints/HintsDispatcher.java | 52 +-- .../apache/cassandra/hints/HintsService.java | 12 +- .../hints/HintsServiceDiagnostics.java | 65 ++++ .../cassandra/hints/HintsServiceEvent.java | 71 ++++ .../cassandra/locator/TokenMetadata.java | 2 + .../locator/TokenMetadataDiagnostics.java | 46 +++ .../cassandra/locator/TokenMetadataEvent.java | 62 ++++ .../org/apache/cassandra/schema/Diff.java | 5 + .../cassandra/schema/MigrationManager.java | 33 +- .../cassandra/schema/MigrationTask.java | 5 + .../org/apache/cassandra/schema/Schema.java | 34 ++ .../schema/SchemaAnnouncementDiagnostics.java | 60 ++++ .../schema/SchemaAnnouncementEvent.java | 104 ++++++ .../cassandra/schema/SchemaDiagnostics.java | 178 ++++++++++ .../apache/cassandra/schema/SchemaEvent.java | 318 ++++++++++++++++++ .../schema/SchemaMigrationDiagnostics.java | 83 +++++ .../schema/SchemaMigrationEvent.java | 114 +++++++ .../schema/SchemaPushVerbHandler.java | 1 + .../PendingRangeCalculatorService.java | 23 +- ...dingRangeCalculatorServiceDiagnostics.java | 73 ++++ .../PendingRangeCalculatorServiceEvent.java | 69 ++++ .../diag/DiagnosticEventServiceTest.java | 244 ++++++++++++++ 63 files changed, 3047 insertions(+), 40 deletions(-) create mode 100644 src/java/org/apache/cassandra/dht/BootstrapDiagnostics.java create mode 100644 src/java/org/apache/cassandra/dht/BootstrapEvent.java create mode 100644 src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorDiagnostics.java create mode 100644 src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorEvent.java create mode 100644 src/java/org/apache/cassandra/diag/DiagnosticEvent.java create mode 100644 src/java/org/apache/cassandra/diag/DiagnosticEventService.java create mode 100644 src/java/org/apache/cassandra/gms/GossiperDiagnostics.java create mode 100644 src/java/org/apache/cassandra/gms/GossiperEvent.java create mode 100644 src/java/org/apache/cassandra/hints/HintDiagnostics.java create mode 100644 src/java/org/apache/cassandra/hints/HintEvent.java create mode 100644 src/java/org/apache/cassandra/hints/HintsServiceDiagnostics.java create mode 100644 src/java/org/apache/cassandra/hints/HintsServiceEvent.java create mode 100644 src/java/org/apache/cassandra/locator/TokenMetadataDiagnostics.java create mode 100644 src/java/org/apache/cassandra/locator/TokenMetadataEvent.java create mode 100644 src/java/org/apache/cassandra/schema/SchemaAnnouncementDiagnostics.java create mode 100644 src/java/org/apache/cassandra/schema/SchemaAnnouncementEvent.java create mode 100644 src/java/org/apache/cassandra/schema/SchemaDiagnostics.java create mode 100644 src/java/org/apache/cassandra/schema/SchemaEvent.java create mode 100644 src/java/org/apache/cassandra/schema/SchemaMigrationDiagnostics.java create mode 100644 src/java/org/apache/cassandra/schema/SchemaMigrationEvent.java create mode 100644 src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceDiagnostics.java create mode 100644 src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceEvent.java create mode 100644 test/unit/org/apache/cassandra/diag/DiagnosticEventServiceTest.java diff --git a/CHANGES.txt b/CHANGES.txt index d8aca566ba..ceba8438cd 100644 --- a/CHANGES.txt +++ b/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) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 663daaada6..c7c27851f3 100644 --- a/conf/cassandra.yaml +++ b/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 diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 3a7ff0dcb4..46dbc1c93a 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/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()} */ diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 366dac7633..65a34f0186 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/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) { diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java index c2d0e4c46f..12e73d0db4 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java index a081a2c3e1..3ec75b25e7 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java +++ b/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); /** diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTypeStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTypeStatement.java index 50f09a0fd9..6eab2bada0 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTypeStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterViewStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterViewStatement.java index 2ecc0952a0..1931bb489d 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterViewStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateAggregateStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateAggregateStatement.java index cd9808a526..462623d5aa 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateAggregateStatement.java +++ b/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))); diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java index 13e173f47e..20c4ad902b 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java index 0065a4c5b0..df41358d9f 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java index ecd19edc7f..f85a4e94b9 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java index ff26f0dd0f..62fcafe6b8 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java +++ b/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(); diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTriggerStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTriggerStatement.java index cb6d14e4f2..e85ffd80ae 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTriggerStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTypeStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTypeStatement.java index c328eb7180..7c1717e24c 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTypeStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java index f97b0fe3f6..5f6200179f 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropAggregateStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropAggregateStatement.java index 564f267660..d24f77e6c4 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropAggregateStatement.java +++ b/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> prepareArgumentTypes(Types types) { return arguments.stream() diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropFunctionStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropFunctionStatement.java index 9433833bf5..f7d7d4a6b9 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropFunctionStatement.java +++ b/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> prepareArgumentTypes(Types types) { return arguments.stream() diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java index 6a6f8d933f..2186470a31 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java index ae5cf06db5..f2bd30b249 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java index 9be59af9df..15c2a03a3a 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTriggerStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTriggerStatement.java index 8de47c2a68..967e56834f 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTriggerStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTypeStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTypeStatement.java index d51954c88e..6cda7ba6a0 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTypeStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropViewStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropViewStatement.java index 807d03dfc6..2c73717546 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropViewStatement.java +++ b/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; diff --git a/src/java/org/apache/cassandra/dht/BootStrapper.java b/src/java/org/apache/cassandra/dht/BootStrapper.java index 432586b13b..92bf8c85b9 100644 --- a/src/java/org/apache/cassandra/dht/BootStrapper.java +++ b/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 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 tokens = getRandomTokens(metadata, numTokens); + BootstrapDiagnostics.useRandomTokens(address, metadata, numTokens, tokens); + return tokens; } private static Collection 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 tokens = TokenAllocation.allocateTokens(metadata, rs, address, numTokens); + BootstrapDiagnostics.tokensAllocated(address, metadata, allocationKeyspace, numTokens, tokens); + return tokens; } public static Collection getRandomTokens(TokenMetadata metadata, int numTokens) diff --git a/src/java/org/apache/cassandra/dht/BootstrapDiagnostics.java b/src/java/org/apache/cassandra/dht/BootstrapDiagnostics.java new file mode 100644 index 0000000000..56955326de --- /dev/null +++ b/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 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 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 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); + } + +} diff --git a/src/java/org/apache/cassandra/dht/BootstrapEvent.java b/src/java/org/apache/cassandra/dht/BootstrapEvent.java new file mode 100644 index 0000000000..5bad09a19f --- /dev/null +++ b/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 tokens; + + BootstrapEvent(BootstrapEventType type, InetAddressAndPort address, @Nullable TokenMetadata tokenMetadata, + @Nullable String allocationKeyspace, int numTokens, ImmutableCollection 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 toMap() + { + // be extra defensive against nulls and bugs + HashMap 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; + } +} diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java b/src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java index 54d80dcd19..f6a159208b 100644 --- a/src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java +++ b/src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java @@ -86,6 +86,7 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase sortedUnits.add(new Weighted(unitInfo.ownership, unitInfo)); } + TokenAllocatorDiagnostics.tokenInfosCreated(this, sortedUnits, sortedTokens, first); return first; } @@ -127,6 +128,7 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase } 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 extends TokenAllocatorBase } sortedUnits.add(new Weighted<>(newUnitInfo.ownership, newUnitInfo)); + TokenAllocatorDiagnostics.unitedAdded(this, numTokens, sortedUnits, sortedTokens, newTokens, newUnit); return newTokens; } @@ -257,6 +260,7 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase tokens.add(tokenInfo.value.token); } sortedTokens.keySet().removeAll(tokens); + TokenAllocatorDiagnostics.unitRemoved(this, n, sortedUnits, sortedTokens); } public int getReplicas() diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java b/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java index 87dba59129..efd2766aa6 100644 --- a/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java +++ b/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java @@ -132,7 +132,9 @@ class ReplicationAwareTokenAllocator extends TokenAllocatorBase } } - return ImmutableList.copyOf(unitToTokens.get(newUnit)); + ImmutableList newTokens = ImmutableList.copyOf(unitToTokens.get(newUnit)); + TokenAllocatorDiagnostics.unitedAdded(this, numTokens, unitToTokens, sortedTokens, newTokens, newUnit); + return newTokens; } private Collection generateRandomTokens(Unit newUnit, int numTokens) @@ -148,6 +150,7 @@ class ReplicationAwareTokenAllocator extends TokenAllocatorBase unitToTokens.put(newUnit, token); } } + TokenAllocatorDiagnostics.randomTokensGenerated(this, numTokens, unitToTokens, sortedTokens, newUnit, tokens); return tokens; } @@ -176,6 +179,7 @@ class ReplicationAwareTokenAllocator extends TokenAllocatorBase curr = curr.next; } while (curr != first); + TokenAllocatorDiagnostics.tokenInfosCreated(this, unitToTokens, first); return first; } @@ -526,6 +530,7 @@ class ReplicationAwareTokenAllocator extends TokenAllocatorBase { Collection tokens = unitToTokens.removeAll(n); sortedTokens.keySet().removeAll(tokens); + TokenAllocatorDiagnostics.unitRemoved(this, n, unitToTokens, sortedTokens); } public int unitCount() diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorDiagnostics.java b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorDiagnostics.java new file mode 100644 index 0000000000..cc27a475cc --- /dev/null +++ b/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 void noReplicationTokenAllocatorInstanciated(NoReplicationTokenAllocator 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 void replicationTokenAllocatorInstanciated(ReplicationAwareTokenAllocator 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 void unitedAdded(TokenAllocatorBase allocator, int numTokens, + Queue> sortedUnits, NavigableMap sortedTokens, + List 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 void unitedAdded(TokenAllocatorBase allocator, int numTokens, + Multimap unitToTokens, NavigableMap sortedTokens, + List 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 void unitRemoved(TokenAllocatorBase allocator, Unit unit, + Queue> sortedUnits, Map 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 void unitRemoved(TokenAllocatorBase allocator, Unit unit, + Multimap unitToTokens, Map 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 void tokenInfosCreated(TokenAllocatorBase allocator, Queue> sortedUnits, + Map sortedTokens, TokenInfo 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 void tokenInfosCreated(TokenAllocatorBase allocator, Multimap unitToTokens, + TokenInfo 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 void randomTokensGenerated(TokenAllocatorBase allocator, + int numTokens, Queue> sortedUnits, + NavigableMap sortedTokens, Unit newUnit, + Set 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 void randomTokensGenerated(TokenAllocatorBase allocator, + int numTokens, Multimap unitToTokens, + NavigableMap sortedTokens, Unit newUnit, + Set 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); + } + +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorEvent.java b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorEvent.java new file mode 100644 index 0000000000..ca59938678 --- /dev/null +++ b/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 extends DiagnosticEvent +{ + + private final TokenAllocatorEventType type; + private final TokenAllocatorBase allocator; + private final int replicas; + @Nullable + private final Integer numTokens; + @Nullable + private final Collection> sortedUnits; + @Nullable + private final Map> unitToTokens; + @Nullable + private final ImmutableMap sortedTokens; + @Nullable + private final List tokens; + @Nullable + private final Unit unit; + @Nullable + private final TokenInfo tokenInfo; + + TokenAllocatorEvent(TokenAllocatorEventType type, TokenAllocatorBase allocator, @Nullable Integer numTokens, + @Nullable ImmutableList> sortedUnits, @Nullable ImmutableMap> unitToTokens, + @Nullable ImmutableMap sortedTokens, @Nullable ImmutableList tokens, Unit unit, + @Nullable TokenInfo 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 toMap() + { + // be extra defensive against nulls and bugs + HashMap 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; + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorFactory.java b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorFactory.java index 5fdba02714..117fd09f5f 100644 --- a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorFactory.java +++ b/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 allocator = new NoReplicationTokenAllocator<>(sortedTokens, strategy, partitioner); + TokenAllocatorDiagnostics.noReplicationTokenAllocatorInstanciated(allocator); + return allocator; } logger.info("Using ReplicationAwareTokenAllocator."); - return new ReplicationAwareTokenAllocator<>(sortedTokens, strategy, partitioner); + ReplicationAwareTokenAllocator allocator = new ReplicationAwareTokenAllocator<>(sortedTokens, strategy, partitioner); + TokenAllocatorDiagnostics.replicationTokenAllocatorInstanciated(allocator); + return allocator; } } diff --git a/src/java/org/apache/cassandra/diag/DiagnosticEvent.java b/src/java/org/apache/cassandra/diag/DiagnosticEvent.java new file mode 100644 index 0000000000..5de703bd71 --- /dev/null +++ b/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 toMap(); +} diff --git a/src/java/org/apache/cassandra/diag/DiagnosticEventService.java b/src/java/org/apache/cassandra/diag/DiagnosticEventService.java new file mode 100644 index 0000000000..577d0ba026 --- /dev/null +++ b/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> subscribersAll = ImmutableSet.of(); + + // Subscribers for particular event class, e.g. BootstrapEvent + private ImmutableSetMultimap, Consumer> subscribersByClass = ImmutableSetMultimap.of(); + + // Subscribers for event class and type, e.g. BootstrapEvent#TOKENS_ALLOCATED + private ImmutableMap, Consumer>> 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, Consumer> consumersByType = subscribersByClassAndType.get(event.getClass()); + if (consumersByType != null) + { + ImmutableCollection> consumers = consumersByType.get(event.getType()); + if (consumers != null) + { + for (Consumer consumer : consumers) + consumer.accept(event); + } + } + + // event class + Set> consumersByEvents = subscribersByClass.get(event.getClass()); + if (consumersByEvents != null) + { + for (Consumer consumer : consumersByEvents) + consumer.accept(event); + } + + // all events + for (Consumer 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 void subscribe(Class event, Consumer consumer) + { + subscribersByClass = ImmutableSetMultimap., Consumer>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 > void subscribe(Class event, + T eventType, + Consumer consumer) + { + ImmutableSetMultimap.Builder, Consumer> byTypeBuilder = ImmutableSetMultimap.builder(); + if (subscribersByClassAndType.containsKey(event)) + byTypeBuilder.putAll(subscribersByClassAndType.get(event)); + byTypeBuilder.put(eventType, new TypedConsumerWrapper<>(consumer)); + + ImmutableMap.Builder, Consumer>> 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 consumer) + { + subscribersAll = ImmutableSet.>builder() + .addAll(subscribersAll) + .add(consumer) + .build(); + } + + /** + * De-registers event handler from receiving any further events. + * @param consumer Consumer registered for receiving events + */ + public synchronized void unsubscribe(Consumer consumer) + { + // all events + subscribersAll = ImmutableSet.copyOf(Iterables.filter(subscribersAll, (c) -> c != consumer)); + + // event class + ImmutableSetMultimap.Builder, Consumer> byClassBuilder = ImmutableSetMultimap.builder(); + Collection, Consumer>> entries = subscribersByClass.entries(); + for (Map.Entry, Consumer> entry : entries) + { + Consumer 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, Consumer>> byClassAndTypeBuilder = ImmutableMap.builder(); + for (Map.Entry, Consumer>> byClassEntry : subscribersByClassAndType.entrySet()) + { + ImmutableSetMultimap.Builder, Consumer> byTypeBuilder = ImmutableSetMultimap.builder(); + ImmutableSetMultimap, Consumer> byTypeConsumers = byClassEntry.getValue(); + Iterables.filter(byTypeConsumers.entries(), (e) -> + { + if (e == null || e.getValue() == null) return false; + Consumer subscriber = e.getValue(); + if (subscriber instanceof TypedConsumerWrapper) + subscriber = ((TypedConsumerWrapper) subscriber).wrapped; + return subscriber != consumer; + }).forEach(byTypeBuilder::put); + + ImmutableSetMultimap, Consumer> 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 boolean hasSubscribers(Class 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 > boolean hasSubscribers(Class event, T eventType) + { + if (!subscribersAll.isEmpty()) + return true; + + ImmutableSet> subscribers = subscribersByClass.get(event); + if (subscribers != null && !subscribers.isEmpty()) + return true; + + ImmutableSetMultimap, Consumer> byType = subscribersByClassAndType.get(event); + if (byType == null || byType.isEmpty()) return false; + + Set> 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 boolean isEnabled(Class 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 > boolean isEnabled(Class 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 implements Consumer + { + private final Consumer wrapped; + + private TypedConsumerWrapper(Consumer 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); + } + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 7bb2583647..170843bcd8 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/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 getEndpointStateMap() + { + return ImmutableMap.copyOf(endpointStateMap); + } + + Map getJustRemovedEndpoints() + { + return ImmutableMap.copyOf(justRemovedEndpoints); + } + + Map getUnreachableEndpoints() + { + return ImmutableMap.copyOf(unreachableEndpoints); + } + + Set 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) diff --git a/src/java/org/apache/cassandra/gms/GossiperDiagnostics.java b/src/java/org/apache/cassandra/gms/GossiperDiagnostics.java new file mode 100644 index 0000000000..57552cca5a --- /dev/null +++ b/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); + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/gms/GossiperEvent.java b/src/java/org/apache/cassandra/gms/GossiperEvent.java new file mode 100644 index 0000000000..2de88bccff --- /dev/null +++ b/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 endpointStateMap; + private final boolean inShadowRound; + private final Map justRemovedEndpoints; + private final long lastProcessedMessageAt; + private final Set liveEndpoints; + private final List seeds; + private final Set seedsInShadowRound; + private final Map 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 getType() + { + return type; + } + + public HashMap toMap() + { + // be extra defensive against nulls and bugs + HashMap 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; + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/hints/Hint.java b/src/java/org/apache/cassandra/hints/Hint.java index b0abd501f0..7e4618cfdc 100644 --- a/src/java/org/apache/cassandra/hints/Hint.java +++ b/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()); } diff --git a/src/java/org/apache/cassandra/hints/HintDiagnostics.java b/src/java/org/apache/cassandra/hints/HintDiagnostics.java new file mode 100644 index 0000000000..3ff0834c7f --- /dev/null +++ b/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); + } + +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/hints/HintEvent.java b/src/java/org/apache/cassandra/hints/HintEvent.java new file mode 100644 index 0000000000..011f248f09 --- /dev/null +++ b/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 getType() + { + return type; + } + + public HashMap toMap() + { + // be extra defensive against nulls and bugs + HashMap 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; + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java b/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java index 92dbc2984c..2d9fd9d678 100644 --- a/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java +++ b/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(); + } } diff --git a/src/java/org/apache/cassandra/hints/HintsDispatcher.java b/src/java/org/apache/cassandra/hints/HintsDispatcher.java index d0d9aace2a..2cff18608c 100644 --- a/src/java/org/apache/cassandra/hints/HintsDispatcher.java +++ b/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; diff --git a/src/java/org/apache/cassandra/hints/HintsService.java b/src/java/org/apache/cassandra/hints/HintsService.java index a46eb5221b..0cd1278e5d 100644 --- a/src/java/org/apache/cassandra/hints/HintsService.java +++ b/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); } /** diff --git a/src/java/org/apache/cassandra/hints/HintsServiceDiagnostics.java b/src/java/org/apache/cassandra/hints/HintsServiceDiagnostics.java new file mode 100644 index 0000000000..f4cf14930c --- /dev/null +++ b/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); + } + +} diff --git a/src/java/org/apache/cassandra/hints/HintsServiceEvent.java b/src/java/org/apache/cassandra/hints/HintsServiceEvent.java new file mode 100644 index 0000000000..72497a0022 --- /dev/null +++ b/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 getType() + { + return type; + } + + public HashMap toMap() + { + // be extra defensive against nulls and bugs + HashMap ret = new HashMap<>(); + ret.put("isDispatchPaused", isDispatchPaused); + ret.put("isShutdown", isShutdown); + ret.put("dispatchExecutorIsPaused", dispatchExecutorIsPaused); + ret.put("dispatchExecutorHasScheduledDispatches", dispatchExecutorHasScheduledDispatches); + return ret; + } +} diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index e2c4628642..46c191f60c 100644 --- a/src/java/org/apache/cassandra/locator/TokenMetadata.java +++ b/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()) diff --git a/src/java/org/apache/cassandra/locator/TokenMetadataDiagnostics.java b/src/java/org/apache/cassandra/locator/TokenMetadataDiagnostics.java new file mode 100644 index 0000000000..0221f1e22f --- /dev/null +++ b/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); + } + +} diff --git a/src/java/org/apache/cassandra/locator/TokenMetadataEvent.java b/src/java/org/apache/cassandra/locator/TokenMetadataEvent.java new file mode 100644 index 0000000000..c3ed074b33 --- /dev/null +++ b/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 toMap() + { + // be extra defensive against nulls and bugs + HashMap ret = new HashMap<>(); + ret.put("keyspace", keyspace); + ret.put("tokenMetadata", tokenMetadata.toString()); + return ret; + } +} diff --git a/src/java/org/apache/cassandra/schema/Diff.java b/src/java/org/apache/cassandra/schema/Diff.java index 36c0687a61..7112c85ba2 100644 --- a/src/java/org/apache/cassandra/schema/Diff.java +++ b/src/java/org/apache/cassandra/schema/Diff.java @@ -55,5 +55,10 @@ public class Diff this.after = after; this.kind = kind; } + + public String toString() + { + return String.format("%s -> %s (%s)", before, after, kind); + } } } diff --git a/src/java/org/apache/cassandra/schema/MigrationManager.java b/src/java/org/apache/cassandra/schema/MigrationManager.java index ac95054d1e..a439e2eb08 100644 --- a/src/java/org/apache/cassandra/schema/MigrationManager.java +++ b/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 schemaDestinationEndpoints = new HashSet<>(); + Set 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 schemaDestinationEndpoints = new HashSet<>(); + Set 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..."); diff --git a/src/java/org/apache/cassandra/schema/MigrationTask.java b/src/java/org/apache/cassandra/schema/MigrationTask.java index 6ff206a6ba..bf96fb27b9 100644 --- a/src/java/org/apache/cassandra/schema/MigrationTask.java +++ b/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 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); } } diff --git a/src/java/org/apache/cassandra/schema/Schema.java b/src/java/org/apache/cassandra/schema/Schema.java index e1353cd7dd..970d9ac2a9 100644 --- a/src/java/org/apache/cassandra/schema/Schema.java +++ b/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, 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 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) diff --git a/src/java/org/apache/cassandra/schema/SchemaAnnouncementDiagnostics.java b/src/java/org/apache/cassandra/schema/SchemaAnnouncementDiagnostics.java new file mode 100644 index 0000000000..be60b1b5d8 --- /dev/null +++ b/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 schemaDestinationEndpoints, Set 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 schemaDestinationEndpoints, Set 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); + } +} diff --git a/src/java/org/apache/cassandra/schema/SchemaAnnouncementEvent.java b/src/java/org/apache/cassandra/schema/SchemaAnnouncementEvent.java new file mode 100644 index 0000000000..4e0bd68716 --- /dev/null +++ b/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 schemaDestinationEndpoints; + @Nullable + private final Set 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 schemaDestinationEndpoints, + @Nullable Set 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 toMap() + { + HashMap ret = new HashMap<>(); + if (schemaDestinationEndpoints != null) + { + Set eps = schemaDestinationEndpoints.stream().map(InetAddressAndPort::toString).collect(Collectors.toSet()); + ret.put("endpointDestinations", new HashSet<>(eps)); + } + if (schemaEndpointsIgnored != null) + { + Set 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 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; + } +} diff --git a/src/java/org/apache/cassandra/schema/SchemaDiagnostics.java b/src/java/org/apache/cassandra/schema/SchemaDiagnostics.java new file mode 100644 index 0000000000..12b8409ebc --- /dev/null +++ b/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 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); + } + +} diff --git a/src/java/org/apache/cassandra/schema/SchemaEvent.java b/src/java/org/apache/cassandra/schema/SchemaEvent.java new file mode 100644 index 0000000000..e26cee5b45 --- /dev/null +++ b/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 keyspaces; + private final HashMap indexTables; + private final HashMap tables; + private final ArrayList nonSystemKeyspaces; + private final ArrayList 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 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 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, TableMetadataRef> indexTableMetadataRefs = schema.getIndexTableMetadataRefs(); + Map 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 tableMetadataRefs = schema.getTableMetadataRefs(); + Map 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 toMap() + { + HashMap 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 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 repr(Diff diff) + { + HashMap 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 repr(KeyspaceMetadata ksm) + { + HashMap 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 repr(TableMetadata table) + { + HashMap 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 repr(TableParams params) + { + HashMap 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 repr(CachingParams caching) + { + HashMap ret = new HashMap<>(); + if (caching == null) return ret; + ret.putAll(caching.asMap()); + return ret; + } + + private HashMap repr(CompactionParams comp) + { + HashMap ret = new HashMap<>(); + if (comp == null) return ret; + ret.putAll(comp.asMap()); + return ret; + } + + private HashMap repr(CompressionParams compr) + { + HashMap ret = new HashMap<>(); + if (compr == null) return ret; + ret.putAll(compr.asMap()); + return ret; + } + + private HashMap repr(IndexMetadata index) + { + HashMap 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> repr(Triggers triggers) + { + List> ret = new ArrayList<>(); + if (triggers == null) return ret; + Iterator iter = triggers.iterator(); + while (iter.hasNext()) ret.add(repr(iter.next())); + return ret; + } + + private HashMap repr(TriggerMetadata trigger) + { + HashMap ret = new HashMap<>(); + if (trigger == null) return ret; + ret.put("name", trigger.name); + ret.put("classOption", trigger.classOption); + return ret; + } + + private HashMap repr(ColumnMetadata col) + { + HashMap 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 repr(DroppedColumn column) + { + HashMap ret = new HashMap<>(); + if (column == null) return ret; + ret.put("droppedTime", column.droppedTime); + ret.put("column", repr(column.column)); + return ret; + } +} diff --git a/src/java/org/apache/cassandra/schema/SchemaMigrationDiagnostics.java b/src/java/org/apache/cassandra/schema/SchemaMigrationDiagnostics.java new file mode 100644 index 0000000000..62f1768d89 --- /dev/null +++ b/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); + } +} diff --git a/src/java/org/apache/cassandra/schema/SchemaMigrationEvent.java b/src/java/org/apache/cassandra/schema/SchemaMigrationEvent.java new file mode 100644 index 0000000000..2c17235898 --- /dev/null +++ b/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 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 toMap() + { + HashMap 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; + } +} diff --git a/src/java/org/apache/cassandra/schema/SchemaPushVerbHandler.java b/src/java/org/apache/cassandra/schema/SchemaPushVerbHandler.java index f939cda117..358739a1b5 100644 --- a/src/java/org/apache/cassandra/schema/SchemaPushVerbHandler.java +++ b/src/java/org/apache/cassandra/schema/SchemaPushVerbHandler.java @@ -42,6 +42,7 @@ public final class SchemaPushVerbHandler implements IVerbHandler Schema.instance.mergeAndAnnounceVersion(message.payload)); } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java b/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java index 297774aa03..e1c0f550c5 100644 --- a/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java +++ b/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 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() diff --git a/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceDiagnostics.java b/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceDiagnostics.java new file mode 100644 index 0000000000..ec09e3fdf4 --- /dev/null +++ b/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); + } +} diff --git a/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceEvent.java b/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceEvent.java new file mode 100644 index 0000000000..3024149b16 --- /dev/null +++ b/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 toMap() + { + // be extra defensive against nulls and bugs + HashMap ret = new HashMap<>(); + ret.put("taskCount", taskCount); + return ret; + } +} diff --git a/test/unit/org/apache/cassandra/diag/DiagnosticEventServiceTest.java b/test/unit/org/apache/cassandra/diag/DiagnosticEventServiceTest.java new file mode 100644 index 0000000000..0db5ef68b0 --- /dev/null +++ b/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 consumer1 = (event) -> + { + }; + Consumer consumer2 = (event) -> + { + }; + Consumer 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 consumer1 = (event) -> + { + }; + Consumer consumer2 = (event) -> + { + }; + Consumer 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 consumerAll1 = (event) -> + { + }; + Consumer consumerAll2 = (event) -> + { + }; + Consumer 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 consumer = (event) -> + { + }; + instance.subscribe(TestEvent1.class, consumer); + Consumer 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 events = ImmutableList.of(a, b, c, c, c); + + List consumed = new LinkedList<>(); + Consumer consumer = consumed::add; + Consumer 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 toMap() + { + return null; + } + } + + private static class TestEvent2 extends DiagnosticEvent + { + public TestEventType getType() + { + return TestEventType.TEST2; + } + + public HashMap toMap() + { + return null; + } + } + + private enum TestEventType { TEST1, TEST2, TEST3 }; +}