-
Notifications
You must be signed in to change notification settings - Fork 39
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Avoid calling RegularStatement.toString() to de-duplicate the insert …
…statements and introduce additional options to control what gets indexed.
- Loading branch information
Jesse White
committed
Jun 2, 2016
1 parent
787bd4e
commit b6eebfd
Showing
8 changed files
with
529 additions
and
100 deletions.
There are no files selected for viewing
308 changes: 217 additions & 91 deletions
308
cassandra/search/src/main/java/org/opennms/newts/cassandra/search/CassandraIndexer.java
Large diffs are not rendered by default.
Oops, something went wrong.
87 changes: 87 additions & 0 deletions
87
...dra/search/src/main/java/org/opennms/newts/cassandra/search/CassandraIndexingOptions.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| /* | ||
| * Copyright 2016, The OpenNMS Group | ||
| * | ||
| * Licensed 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.opennms.newts.cassandra.search; | ||
|
|
||
| import javax.inject.Inject; | ||
| import javax.inject.Named; | ||
|
|
||
| public class CassandraIndexingOptions { | ||
|
|
||
| private final boolean m_enableHierarchicalIndexing; | ||
| private final boolean m_indexUsingDefaultTerm; | ||
| private final boolean m_indexResourceTerms; | ||
|
|
||
| private final int m_maxBatchSize; | ||
|
|
||
| public static class Builder { | ||
| private int maxBatchSize = 16; | ||
| private boolean enableHierarchicalIndexing = true; | ||
| private boolean indexUsingDefaultTerm = true; | ||
| private boolean indexResourceTerms = true; | ||
|
|
||
| public Builder withHierarchicalIndexing(boolean enableHierarchicalIndexing) { | ||
| this.enableHierarchicalIndexing = enableHierarchicalIndexing; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder withIndexUsingDefaultTerm(boolean indexUsingDefaultTerm) { | ||
| this.indexUsingDefaultTerm = indexUsingDefaultTerm; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder withIndexResourceTerms(boolean indexResourceTerms) { | ||
| this.indexResourceTerms = indexResourceTerms; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder withMaxBatchSize(int maxBatchSize) { | ||
| this.maxBatchSize = maxBatchSize; | ||
| return this; | ||
| } | ||
|
|
||
| public CassandraIndexingOptions build() { | ||
| return new CassandraIndexingOptions(this); | ||
| } | ||
| } | ||
|
|
||
| public CassandraIndexingOptions(CassandraIndexingOptions.Builder builder) { | ||
| m_maxBatchSize = builder.maxBatchSize; | ||
| m_enableHierarchicalIndexing = builder.enableHierarchicalIndexing; | ||
| m_indexUsingDefaultTerm = builder.indexUsingDefaultTerm; | ||
| m_indexResourceTerms = builder.indexResourceTerms; | ||
| } | ||
|
|
||
| @Inject | ||
| public CassandraIndexingOptions(@Named("search.hierarical-indexing") boolean enableHierarchicalIndexing) { | ||
| this(new Builder().withHierarchicalIndexing(enableHierarchicalIndexing)); | ||
| } | ||
|
|
||
| public boolean isHierarchicalIndexingEnabled() { | ||
| return m_enableHierarchicalIndexing; | ||
| } | ||
|
|
||
| public boolean shouldIndexUsingDefaultTerm() { | ||
| return m_indexUsingDefaultTerm; | ||
| } | ||
|
|
||
| public boolean shouldIndexResourceTerms() { | ||
| return m_indexResourceTerms; | ||
| } | ||
|
|
||
| public int getMaxBatchSize() { | ||
| return m_maxBatchSize; | ||
| } | ||
| } |
50 changes: 50 additions & 0 deletions
50
...a/search/src/main/java/org/opennms/newts/cassandra/search/support/StatementGenerator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| /* | ||
| * Copyright 2016, The OpenNMS Group | ||
| * | ||
| * Licensed 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.opennms.newts.cassandra.search.support; | ||
|
|
||
| import org.opennms.newts.api.Context; | ||
|
|
||
| import com.datastax.driver.core.Statement; | ||
|
|
||
| /** | ||
| * Used to group and de-duplicate statements before they are generated and executed. | ||
| * | ||
| * @author jwhite | ||
| */ | ||
| public interface StatementGenerator { | ||
|
|
||
| /** | ||
| * Returns a key which can be used to group statements into batches, or null | ||
| * if the statement should never be be batched. | ||
| * | ||
| * @return key or null | ||
| */ | ||
| String getKey(); | ||
|
|
||
| /** | ||
| * Returns the context with this this statement is associated. | ||
| * | ||
| * @return context | ||
| */ | ||
| Context getContext(); | ||
|
|
||
| /** | ||
| * Generates the statement. | ||
| * | ||
| * @return statement | ||
| */ | ||
| Statement toStatement(); | ||
| } |
51 changes: 51 additions & 0 deletions
51
...andra/search/src/main/java/org/opennms/newts/cassandra/search/support/StatementUtils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| package org.opennms.newts.cassandra.search.support; | ||
|
|
||
| import static com.datastax.driver.core.querybuilder.QueryBuilder.unloggedBatch; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
|
|
||
| import org.opennms.newts.cassandra.ContextConfigurations; | ||
|
|
||
| import com.datastax.driver.core.RegularStatement; | ||
| import com.datastax.driver.core.Statement; | ||
| import com.google.common.collect.Lists; | ||
| import com.google.common.collect.Maps; | ||
|
|
||
| public class StatementUtils { | ||
|
|
||
| public static List<Statement> getStatements(ContextConfigurations contextConfigurations, int maxBatchSize, | ||
| Set<StatementGenerator> generators) { | ||
| List<Statement> statementsToExecute = Lists.newArrayList(); | ||
|
|
||
| Map<String, List<Statement>> statementsByKey = Maps.newHashMap(); | ||
| for (StatementGenerator generator : generators) { | ||
| Statement statement = generator.toStatement() | ||
| .setConsistencyLevel(contextConfigurations.getWriteConsistency(generator.getContext())); | ||
| String key = generator.getKey(); | ||
| if (key == null) { | ||
| // Don't try batching these | ||
| statementsToExecute.add(statement); | ||
| continue; | ||
| } | ||
|
|
||
| // Group these by key | ||
| List<Statement> statementsForKey = statementsByKey.get(key); | ||
| if (statementsForKey == null) { | ||
| statementsForKey = Lists.newArrayList(); | ||
| statementsByKey.put(key, statementsForKey); | ||
| } | ||
| statementsForKey.add(statement); | ||
| } | ||
|
|
||
| // Consolidate the grouped statements into batches | ||
| for (List<Statement> statementsForKey: statementsByKey.values()) { | ||
| for (List<Statement> partition : Lists.partition(statementsForKey, maxBatchSize)) { | ||
| statementsToExecute.add(unloggedBatch(partition.toArray(new RegularStatement[partition.size()]))); | ||
| } | ||
| } | ||
|
|
||
| return statementsToExecute; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
86 changes: 86 additions & 0 deletions
86
...search/src/test/java/org/opennms/newts/cassandra/search/CassandraIndexerStressITCase.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| /* | ||
| * Copyright 2016, The OpenNMS Group | ||
| * | ||
| * Licensed 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.opennms.newts.cassandra.search; | ||
|
|
||
| import static org.mockito.Matchers.any; | ||
| import static org.mockito.Mockito.mock; | ||
| import static org.mockito.Mockito.when; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| import org.junit.Test; | ||
| import org.opennms.newts.api.Counter; | ||
| import org.opennms.newts.api.MetricType; | ||
| import org.opennms.newts.api.Resource; | ||
| import org.opennms.newts.api.Sample; | ||
| import org.opennms.newts.api.Timestamp; | ||
| import org.opennms.newts.cassandra.CassandraSession; | ||
| import org.opennms.newts.cassandra.ContextConfigurations; | ||
|
|
||
| import com.codahale.metrics.MetricRegistry; | ||
| import com.datastax.driver.core.ResultSetFuture; | ||
| import com.datastax.driver.core.Statement; | ||
| import com.google.common.collect.Lists; | ||
|
|
||
| public class CassandraIndexerStressITCase { | ||
|
|
||
| @Test | ||
| public void canIndexManyResources() { | ||
| final int numResources = 20000; | ||
| final int numSamplesPerResource = 3; | ||
|
|
||
| // Setup the indexer | ||
| ResultSetFuture future = mock(ResultSetFuture.class); | ||
| CassandraSession session = mock(CassandraSession.class); | ||
| when(session.executeAsync(any(Statement.class))).thenReturn(future); | ||
|
|
||
| ContextConfigurations contexts = new ContextConfigurations(); | ||
| MetricRegistry metrics = new MetricRegistry(); | ||
|
|
||
| CassandraIndexingOptions options = new CassandraIndexingOptions.Builder() | ||
| .withHierarchicalIndexing(true).build(); | ||
|
|
||
| ResourceIdSplitter resourceIdSplitter = new EscapableResourceIdSplitter(); | ||
| GuavaResourceMetadataCache cache = new GuavaResourceMetadataCache(numResources * 2, metrics); | ||
| CassandraIndexer indexer = new CassandraIndexer(session, 0, cache, metrics, options, | ||
| resourceIdSplitter, contexts); | ||
|
|
||
| // Generate the resources and sample sets | ||
| Resource resources[] = new Resource[numResources]; | ||
| List<List<Sample>> sampleSets = Lists.newArrayListWithCapacity(numResources); | ||
| System.out.println("Building sample sets..."); | ||
| for (int i = 0; i < numResources; i++) { | ||
| resources[i] = new Resource(String.format("snmp:%d:eth0-x:ifHcInOctets", i)); | ||
| List<Sample> samples = Lists.newArrayListWithCapacity(numSamplesPerResource); | ||
| for (int j = 0; j < numSamplesPerResource; j++) { | ||
| samples.add(new Sample(Timestamp.now(), resources[i], "y" + j, MetricType.COUNTER, new Counter(i * j))); | ||
| } | ||
| sampleSets.add(samples); | ||
| }; | ||
| System.out.println("Done building sample sets."); | ||
|
|
||
| // Index the resources and associated samples several times over | ||
| for (int k = 0; k < 3; k++) { | ||
| System.out.println("Indexing samples sets..."); | ||
| long start = System.currentTimeMillis(); | ||
| for (List<Sample> sampleSet : sampleSets) { | ||
| indexer.update(sampleSet); | ||
| } | ||
| long elapsed = System.currentTimeMillis() - start; | ||
| System.out.println("Done indexing samples in : " + elapsed + " ms"); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters