Sign Up
Log In
Log In
or
Sign Up
Places
All Projects
Status Monitor
Collapse sidebar
systemsmanagement:Ardana:8:CentOS
openstack-monasca-persister-java
cassandra.patch
Overview
Repositories
Revisions
Requests
Users
Attributes
Meta
File cassandra.patch of Package openstack-monasca-persister-java
commit 89efc8c491b6cc04c1d31057a0c38ac1c93798f0 Author: James Gu <jgu@suse.com> Date: Thu Aug 31 15:17:11 2017 -0700 Status: not upstream, local backport to Pike URL: taken from https://review.openstack.org/#/c/527855/ Add Cassandra db support Added cassandra db plugins in persister, inlcuding Java and python version. Depends-On: I9aa250bab206438f9b98bd211d472c0d8fe12bde Change-Id: I4bbe48f8fe550385de6ed3e14120850a8542c7c9 story: 2001231 task: 5758 diff --git a/java/pom.xml b/java/pom.xml index e7812aa3edc5..b9ea7277a461 100644 --- a/java/pom.xml +++ b/java/pom.xml @@ -49,6 +49,26 @@ <artifactId>monasca-common-influxdb</artifactId> <version>${mon.common.version}</version> </dependency> + <dependency> + <groupId>monasca-common</groupId> + <artifactId>monasca-common-cassandra</artifactId> + <version>${mon.common.version}</version> + </dependency> + <dependency> + <groupId>com.datastax.cassandra</groupId> + <artifactId>cassandra-driver-core</artifactId> + <version>3.1.0</version> + </dependency> + <dependency> + <groupId>com.datastax.cassandra</groupId> + <artifactId>cassandra-driver-mapping</artifactId> + <version>3.1.0</version> + </dependency> + <dependency> + <groupId>com.datastax.cassandra</groupId> + <artifactId>cassandra-driver-extras</artifactId> + <version>3.1.0</version> + </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> @@ -72,6 +92,12 @@ <groupId>io.dropwizard</groupId> <artifactId>dropwizard-core</artifactId> <version>0.7.0</version> + <exclusions> + <exclusion> + <groupId>com.codahale.metrics</groupId> + <artifactId>metrics-core</artifactId> + </exclusion> + </exclusions> </dependency> <dependency> <groupId>io.dropwizard</groupId> @@ -88,6 +114,11 @@ <artifactId>guice-assistedinject</artifactId> <version>3.0</version> </dependency> + <dependency> + <groupId>com.google.guava</groupId> + <artifactId>guava</artifactId> + <version>17.0</version> + </dependency> <dependency> <groupId>org.mockito</groupId> <artifactId>mockito-all</artifactId> diff --git a/java/src/main/java/monasca/persister/PersisterModule.java b/java/src/main/java/monasca/persister/PersisterModule.java index 67e49539a5cc..54e362cf3fca 100644 --- a/java/src/main/java/monasca/persister/PersisterModule.java +++ b/java/src/main/java/monasca/persister/PersisterModule.java @@ -1,6 +1,8 @@ /* * Copyright (c) 2014 Hewlett-Packard Development Company, L.P. * + * Copyright (c) 2017 SUSE LLC. + * * 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 @@ -46,6 +48,9 @@ import monasca.persister.pipeline.event.AlarmStateTransitionHandlerFactory; import monasca.persister.pipeline.event.MetricHandler; import monasca.persister.pipeline.event.MetricHandlerFactory; import monasca.persister.repository.Repo; +import monasca.persister.repository.cassandra.CassandraAlarmRepo; +import monasca.persister.repository.cassandra.CassandraCluster; +import monasca.persister.repository.cassandra.CassandraMetricRepo; import monasca.persister.repository.influxdb.InfluxV9AlarmRepo; import monasca.persister.repository.influxdb.InfluxV9MetricRepo; import monasca.persister.repository.influxdb.InfluxV9RepoWriter; @@ -56,7 +61,7 @@ public class PersisterModule extends AbstractModule { private static final String VERTICA = "vertica"; private static final String INFLUXDB = "influxdb"; - + private static final String CASSANDRA = "cassandra"; private static final String INFLUXDB_V9 = "v9"; private final PersisterConfig config; @@ -168,6 +173,13 @@ public class PersisterModule extends AbstractModule { bind(new TypeLiteral<Repo<AlarmStateTransitionedEvent>> () {}) .to(InfluxV9AlarmRepo.class); + } else if (config.getDatabaseConfiguration().getDatabaseType().equalsIgnoreCase(CASSANDRA)) { + bind(CassandraCluster.class).in(Singleton.class); + + bind(new TypeLiteral<Repo<MetricEnvelope>>() {}).to(CassandraMetricRepo.class); + + bind(new TypeLiteral<Repo<AlarmStateTransitionedEvent>>() {}).to(CassandraAlarmRepo.class); + } else { System.err.println( diff --git a/java/src/main/java/monasca/persister/configuration/PersisterConfig.java b/java/src/main/java/monasca/persister/configuration/PersisterConfig.java index 3abf8ab86a1b..b8d09a6828ec 100644 --- a/java/src/main/java/monasca/persister/configuration/PersisterConfig.java +++ b/java/src/main/java/monasca/persister/configuration/PersisterConfig.java @@ -1,6 +1,8 @@ /* * Copyright (c) 2014 Hewlett-Packard Development Company, L.P. * + * Copyright (c) 2017 SUSE LLC. + * * 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 @@ -19,6 +21,7 @@ package monasca.persister.configuration; import com.fasterxml.jackson.annotation.JsonProperty; +import monasca.common.configuration.CassandraDbConfiguration; import monasca.common.configuration.DatabaseConfiguration; import monasca.common.configuration.InfluxDbConfiguration; import io.dropwizard.Configuration; @@ -97,4 +100,12 @@ public class PersisterConfig extends Configuration { public InfluxDbConfiguration getInfluxDBConfiguration() { return influxDbConfiguration; } + + @Valid + @JsonProperty + private final CassandraDbConfiguration cassandraDbConfiguration = new CassandraDbConfiguration(); + + public CassandraDbConfiguration getCassandraDbConfiguration() { + return cassandraDbConfiguration; + } } diff --git a/java/src/main/java/monasca/persister/configuration/PipelineConfig.java b/java/src/main/java/monasca/persister/configuration/PipelineConfig.java index c6adef159280..b5f95016b737 100644 --- a/java/src/main/java/monasca/persister/configuration/PipelineConfig.java +++ b/java/src/main/java/monasca/persister/configuration/PipelineConfig.java @@ -42,6 +42,17 @@ public class PipelineConfig { @JsonProperty Integer maxBatchTime; + @JsonProperty + Integer commitBatchTime; + + public Integer getCommitBatchTime() { + return commitBatchTime; + } + + public void setCommitBatchTime(Integer commitBatchTime) { + this.commitBatchTime = commitBatchTime; + } + public String getTopic() { return topic; } diff --git a/java/src/main/java/monasca/persister/consumer/KafkaChannel.java b/java/src/main/java/monasca/persister/consumer/KafkaChannel.java index 1c2081107f39..9c863f3d3e17 100644 --- a/java/src/main/java/monasca/persister/consumer/KafkaChannel.java +++ b/java/src/main/java/monasca/persister/consumer/KafkaChannel.java @@ -1,6 +1,8 @@ /* * Copyright (c) 2014 Hewlett-Packard Development Company, L.P. * + * Copyright (c) 2017 SUSE LLC + * * 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 @@ -44,33 +46,50 @@ public class KafkaChannel { private final String topic; private final ConsumerConnector consumerConnector; private final String threadId; + private final int commitBatchtimeInMills; + private long nextCommitTime; + private boolean commitDirty = false; @Inject - public KafkaChannel( - PersisterConfig configuration, - @Assisted PipelineConfig pipelineConfig, + public KafkaChannel(PersisterConfig configuration, @Assisted PipelineConfig pipelineConfig, @Assisted String threadId) { this.topic = pipelineConfig.getTopic(); this.threadId = threadId; - Properties kafkaProperties = - createKafkaProperties(configuration.getKafkaConfig(), pipelineConfig); + this.commitBatchtimeInMills = pipelineConfig.getCommitBatchTime(); + nextCommitTime = System.currentTimeMillis() + commitBatchtimeInMills; + Properties kafkaProperties = createKafkaProperties(configuration.getKafkaConfig(), pipelineConfig); consumerConnector = Consumer.createJavaConsumerConnector(createConsumerConfig(kafkaProperties)); } public final void markRead() { - this.consumerConnector.commitOffsets(); + if (commitBatchtimeInMills <= 0) { + consumerConnector.commitOffsets(); + } else if (nextCommitTime <= System.currentTimeMillis()) { + consumerConnector.commitOffsets(); + nextCommitTime = System.currentTimeMillis() + commitBatchtimeInMills; + commitDirty = false; + } else { + commitDirty = true; + } + } + + public final void markReadIfDirty() { + if (commitDirty) { + this.consumerConnector.commitOffsets(); + commitDirty = false; + } } public KafkaStream<byte[], byte[]> getKafkaStream() { final Map<String, Integer> topicCountMap = new HashMap<>(); topicCountMap.put(this.topic, 1); - Map<String, List<KafkaStream<byte[], byte[]>>> streamMap = - this.consumerConnector.createMessageStreams(topicCountMap); + Map<String, List<KafkaStream<byte[], byte[]>>> streamMap = this.consumerConnector + .createMessageStreams(topicCountMap); List<KafkaStream<byte[], byte[]>> streams = streamMap.values().iterator().next(); if (streams.size() != 1) { - throw new IllegalStateException(String.format( - "Expected only one stream but instead there are %d", streams.size())); + throw new IllegalStateException( + String.format("Expected only one stream but instead there are %d", streams.size())); } return streams.get(0); } @@ -92,32 +111,28 @@ public class KafkaChannel { properties.put("consumer.id", String.format("%s_%s", pipelineConfig.getConsumerId(), this.threadId)); properties.put("socket.timeout.ms", kafkaConfig.getSocketTimeoutMs().toString()); - properties.put("socket.receive.buffer.bytes", kafkaConfig.getSocketReceiveBufferBytes() - .toString()); - properties.put("fetch.message.max.bytes", kafkaConfig.getFetchMessageMaxBytes() - .toString()); + properties.put("socket.receive.buffer.bytes", kafkaConfig.getSocketReceiveBufferBytes().toString()); + properties.put("fetch.message.max.bytes", kafkaConfig.getFetchMessageMaxBytes().toString()); // Set auto commit to false because the persister is going to explicitly commit properties.put("auto.commit.enable", "false"); - properties.put("queued.max.message.chunks", kafkaConfig.getQueuedMaxMessageChunks() - .toString()); + properties.put("queued.max.message.chunks", kafkaConfig.getQueuedMaxMessageChunks().toString()); properties.put("rebalance.max.retries", kafkaConfig.getRebalanceMaxRetries().toString()); properties.put("fetch.min.bytes", kafkaConfig.getFetchMinBytes().toString()); properties.put("fetch.wait.max.ms", kafkaConfig.getFetchWaitMaxMs().toString()); properties.put("rebalance.backoff.ms", kafkaConfig.getRebalanceBackoffMs().toString()); - properties.put("refresh.leader.backoff.ms", kafkaConfig.getRefreshLeaderBackoffMs() - .toString()); + properties.put("refresh.leader.backoff.ms", kafkaConfig.getRefreshLeaderBackoffMs().toString()); properties.put("auto.offset.reset", kafkaConfig.getAutoOffsetReset()); properties.put("consumer.timeout.ms", kafkaConfig.getConsumerTimeoutMs().toString()); properties.put("client.id", String.format("%s_%s", pipelineConfig.getClientId(), threadId)); - properties.put("zookeeper.session.timeout.ms", kafkaConfig - .getZookeeperSessionTimeoutMs().toString()); - properties.put("zookeeper.connection.timeout.ms", kafkaConfig - .getZookeeperConnectionTimeoutMs().toString()); - properties - .put("zookeeper.sync.time.ms", kafkaConfig.getZookeeperSyncTimeMs().toString()); + properties.put("zookeeper.session.timeout.ms", + kafkaConfig.getZookeeperSessionTimeoutMs().toString()); + properties.put("zookeeper.connection.timeout.ms", + kafkaConfig.getZookeeperConnectionTimeoutMs().toString()); + properties.put("zookeeper.sync.time.ms", kafkaConfig.getZookeeperSyncTimeMs().toString()); for (String key : properties.stringPropertyNames()) { - logger.info("[{}]: " + KAFKA_CONFIGURATION + " " + key + " = " + properties.getProperty(key), threadId); + logger.info("[{}]: " + KAFKA_CONFIGURATION + " " + key + " = " + properties.getProperty(key), + threadId); } return properties; diff --git a/java/src/main/java/monasca/persister/consumer/KafkaConsumerRunnableBasic.java b/java/src/main/java/monasca/persister/consumer/KafkaConsumerRunnableBasic.java index 834a87ebb5f5..c5bb5c2a0c85 100644 --- a/java/src/main/java/monasca/persister/consumer/KafkaConsumerRunnableBasic.java +++ b/java/src/main/java/monasca/persister/consumer/KafkaConsumerRunnableBasic.java @@ -1,6 +1,8 @@ /* * Copyright (c) 2014 Hewlett-Packard Development Company, L.P. * + * Copyright (c) 2017 SUSE LLC. + * * 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 @@ -38,14 +40,13 @@ public class KafkaConsumerRunnableBasic<T> implements Runnable { private final String threadId; private final ManagedPipeline<T> pipeline; private volatile boolean stop = false; + private boolean active = false; private ExecutorService executorService; @Inject - public KafkaConsumerRunnableBasic( - @Assisted KafkaChannel kafkaChannel, - @Assisted ManagedPipeline<T> pipeline, - @Assisted String threadId) { + public KafkaConsumerRunnableBasic(@Assisted KafkaChannel kafkaChannel, + @Assisted ManagedPipeline<T> pipeline, @Assisted String threadId) { this.kafkaChannel = kafkaChannel; this.pipeline = pipeline; @@ -67,8 +68,9 @@ public class KafkaConsumerRunnableBasic<T> implements Runnable { } private void markRead() { - - logger.debug("[{}]: marking read", this.threadId); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: marking read", this.threadId); + } this.kafkaChannel.markRead(); @@ -80,12 +82,30 @@ public class KafkaConsumerRunnableBasic<T> implements Runnable { this.stop = true; + int count = 0; + while (active) { + if (count++ >= 20) { + break; + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { + logger.error("interrupted while waiting for the run loop to stop", e); + break; + } + } + + if (!active) { + this.kafkaChannel.markReadIfDirty(); + } } public void run() { logger.info("[{}]: run", this.threadId); + active = true; + final ConsumerIterator<byte[], byte[]> it = kafkaChannel.getKafkaStream().iterator(); logger.debug("[{}]: KafkaChannel has stream iterator", this.threadId); @@ -121,7 +141,9 @@ public class KafkaConsumerRunnableBasic<T> implements Runnable { final String msg = new String(it.next().message()); - logger.debug("[{}]: {}", this.threadId, msg); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: {}", this.threadId, msg); + } publishEvent(msg); @@ -149,22 +171,24 @@ public class KafkaConsumerRunnableBasic<T> implements Runnable { } catch (Throwable e) { - logger.error( - "[{}]: caught fatal exception while publishing msg. Shutting entire persister down " - + "now!", this.threadId, e); + logger + .error("[{}]: caught fatal exception while publishing msg. Shutting entire persister down " + + "now!", this.threadId, e); - logger.error("[{}]: calling shutdown on executor service", this.threadId); - this.executorService.shutdownNow(); + logger.error("[{}]: calling shutdown on executor service", this.threadId); + this.executorService.shutdownNow(); - logger.error("[{}]: shutting down system. calling system.exit(1)", this.threadId); - System.exit(1); + logger.error("[{}]: shutting down system. calling system.exit(1)", this.threadId); + System.exit(1); - } + } } logger.info("[{}]: calling stop on kafka channel", this.threadId); + active = false; + this.kafkaChannel.stop(); logger.debug("[{}]: exiting main run loop", this.threadId); @@ -183,9 +207,10 @@ public class KafkaConsumerRunnableBasic<T> implements Runnable { private boolean isInterrupted() { - if (Thread.currentThread().interrupted()) { - - logger.debug("[{}]: is interrupted. breaking out of run loop", this.threadId); + if (Thread.interrupted()) { + if (logger.isDebugEnabled()) { + logger.debug("[{}]: is interrupted. breaking out of run loop", this.threadId); + } return true; diff --git a/java/src/main/java/monasca/persister/pipeline/event/FlushableHandler.java b/java/src/main/java/monasca/persister/pipeline/event/FlushableHandler.java index 91c4b97dc5c0..f7697607130c 100644 --- a/java/src/main/java/monasca/persister/pipeline/event/FlushableHandler.java +++ b/java/src/main/java/monasca/persister/pipeline/event/FlushableHandler.java @@ -1,6 +1,8 @@ /* * Copyright (c) 2014 Hewlett-Packard Development Company, L.P. * + * Copyright (c) 2017 SUSE LLC + * * 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 @@ -51,28 +53,18 @@ public abstract class FlushableHandler<T> { protected final String handlerName; - protected FlushableHandler( - PipelineConfig configuration, - Environment environment, - String threadId, + protected FlushableHandler(PipelineConfig configuration, Environment environment, String threadId, int batchSize) { this.threadId = threadId; - this.handlerName = - String.format( - "%s[%s]", - this.getClass().getName(), - threadId); + this.handlerName = String.format("%s[%s]", this.getClass().getName(), threadId); - this.processedMeter = - environment.metrics().meter(handlerName + "." + "events-processed-meter"); + this.processedMeter = environment.metrics().meter(handlerName + "." + "events-processed-meter"); - this.flushMeter = - environment.metrics().meter(handlerName + "." + "flush-meter"); + this.flushMeter = environment.metrics().meter(handlerName + "." + "flush-meter"); - this.flushTimer = - environment.metrics().timer(handlerName + "." + "flush-timer"); + this.flushTimer = environment.metrics().timer(handlerName + "." + "flush-timer"); this.secondsBetweenFlushes = configuration.getMaxBatchTime(); @@ -102,7 +94,7 @@ public abstract class FlushableHandler<T> { } else { - return false; + return false; } } @@ -126,20 +118,26 @@ public abstract class FlushableHandler<T> { private boolean isBatchSize() { - logger.debug("[{}]: checking batch size", this.threadId); + if (logger.isDebugEnabled()) { + + logger.debug("[{}]: checking batch size", this.threadId); + + } if (this.msgCount >= this.batchSize) { - logger.debug("[{}]: batch sized {} attained", this.threadId, this.batchSize); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: batch sized {} attained", this.threadId, this.batchSize); + } return true; } else { - logger.debug("[{}]: batch size now at {}, batch size {} not attained", - this.threadId, - this.msgCount, - this.batchSize); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: batch size now at {}, batch size {} not attained", this.threadId, + this.msgCount, this.batchSize); + } return false; @@ -147,28 +145,26 @@ public abstract class FlushableHandler<T> { } private boolean isFlushTime() { - - logger.debug("[{}]: got heartbeat message, checking flush time. flush every {} seconds.", - this.threadId, - this.secondsBetweenFlushes); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: got heartbeat message, checking flush time. flush every {} seconds.", + this.threadId, this.secondsBetweenFlushes); + } long now = System.currentTimeMillis(); - if (this.flushTimeMillis <= now ) { - - logger.debug( - "[{}]: {} ms past flush time. flushing to repository now.", - this.threadId, - now - this.flushTimeMillis); + if (this.flushTimeMillis <= now) { + if (logger.isDebugEnabled()) { + logger.debug("[{}]: {} ms past flush time. flushing to repository now.", this.threadId, + now - this.flushTimeMillis); + } return true; } else { - - logger.debug( - "[{}]: {} ms to next flush time. no need to flush at this time.", - this.threadId, - this.flushTimeMillis - now); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: {} ms to next flush time. no need to flush at this time.", this.threadId, + this.flushTimeMillis - now); + } return false; @@ -176,8 +172,9 @@ public abstract class FlushableHandler<T> { } public int flush() throws RepoException { - - logger.debug("[{}]: flushing", this.threadId); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: flushing", this.threadId); + } Timer.Context context = this.flushTimer.time(); @@ -185,13 +182,15 @@ public abstract class FlushableHandler<T> { context.stop(); - this.flushMeter.mark(); + this.flushMeter.mark(msgFlushCnt); this.flushTimeMillis = System.currentTimeMillis() + this.millisBetweenFlushes; - logger.debug("[{}]: flushed {} msg", this.threadId, msgFlushCnt); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: flushed {} msg", this.threadId, msgFlushCnt); + } - this.msgCount = 0; + this.msgCount -= msgFlushCnt; this.batchCount++; diff --git a/java/src/main/java/monasca/persister/pipeline/event/MetricHandler.java b/java/src/main/java/monasca/persister/pipeline/event/MetricHandler.java index 76ae2697fd23..75fe2f52e679 100644 --- a/java/src/main/java/monasca/persister/pipeline/event/MetricHandler.java +++ b/java/src/main/java/monasca/persister/pipeline/event/MetricHandler.java @@ -1,6 +1,8 @@ /* * Copyright (c) 2014 Hewlett-Packard Development Company, L.P. - * + * + * Copyright (c) 2017 SUSE LLC + * * 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 @@ -38,28 +40,23 @@ import monasca.persister.repository.RepoException; public class MetricHandler extends FlushableHandler<MetricEnvelope[]> { - private static final Logger logger = - LoggerFactory.getLogger(MetricHandler.class); + private static final Logger logger = LoggerFactory.getLogger(MetricHandler.class); private final Repo<MetricEnvelope> metricRepo; private final Counter metricCounter; @Inject - public MetricHandler( - Repo<MetricEnvelope> metricRepo, - Environment environment, - @Assisted PipelineConfig configuration, - @Assisted("threadId") String threadId, + public MetricHandler(Repo<MetricEnvelope> metricRepo, Environment environment, + @Assisted PipelineConfig configuration, @Assisted("threadId") String threadId, @Assisted("batchSize") int batchSize) { super(configuration, environment, threadId, batchSize); this.metricRepo = metricRepo; - this.metricCounter = - environment.metrics() - .counter(this.handlerName + "." + "metrics-added-to-batch-counter"); + this.metricCounter = environment.metrics() + .counter(this.handlerName + "." + "metrics-added-to-batch-counter"); } @@ -89,12 +86,10 @@ public class MetricHandler extends FlushableHandler<MetricEnvelope[]> { } private void processEnvelope(MetricEnvelope metricEnvelope) { - - logger.debug("[{}]: [{}:{}] {}", - this.threadId, - this.getBatchCount(), - this.getMsgCount(), - metricEnvelope); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: [{}:{}] {}", this.threadId, this.getBatchCount(), this.getMsgCount(), + metricEnvelope); + } this.metricRepo.addToBatch(metricEnvelope, this.threadId); @@ -109,8 +104,8 @@ public class MetricHandler extends FlushableHandler<MetricEnvelope[]> { this.objectMapper.enable(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY); - this.objectMapper.setPropertyNamingStrategy( - PropertyNamingStrategy.CAMEL_CASE_TO_LOWER_CASE_WITH_UNDERSCORES); + this.objectMapper + .setPropertyNamingStrategy(PropertyNamingStrategy.CAMEL_CASE_TO_LOWER_CASE_WITH_UNDERSCORES); } diff --git a/java/src/main/java/monasca/persister/repository/vertica/Sha1HashId.java b/java/src/main/java/monasca/persister/repository/Sha1HashId.java similarity index 76% rename from java/src/main/java/monasca/persister/repository/vertica/Sha1HashId.java rename to java/src/main/java/monasca/persister/repository/Sha1HashId.java index 899b073b6868..74c8ecbae156 100644 --- a/java/src/main/java/monasca/persister/repository/vertica/Sha1HashId.java +++ b/java/src/main/java/monasca/persister/repository/Sha1HashId.java @@ -1,63 +1,73 @@ -/* - * Copyright (c) 2014 Hewlett-Packard Development Company, L.P. - * - * 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 monasca.persister.repository.vertica; - -import org.apache.commons.codec.binary.Hex; - -import java.util.Arrays; - -public class Sha1HashId { - private final byte[] sha1Hash; - - public Sha1HashId(byte[] sha1Hash) { - this.sha1Hash = sha1Hash; - } - - @Override - public String toString() { - return "Sha1HashId{" + "sha1Hash=" + Hex.encodeHexString(sha1Hash) + "}"; - } - - @Override - public boolean equals(Object o) { - if (this == o) - return true; - if (!(o instanceof Sha1HashId)) - return false; - - Sha1HashId that = (Sha1HashId) o; - - if (!Arrays.equals(sha1Hash, that.sha1Hash)) - return false; - - return true; - } - - @Override - public int hashCode() { - return Arrays.hashCode(sha1Hash); - } - - public byte[] getSha1Hash() { - return sha1Hash; - } - - public String toHexString() { - return Hex.encodeHexString(sha1Hash); - } -} +/* + * Copyright (c) 2014 Hewlett-Packard Development Company, L.P. + * + * Copyright (c) 2017 SUSE LLC. + * + * 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 monasca.persister.repository; + +import org.apache.commons.codec.binary.Hex; + +import java.nio.ByteBuffer; +import java.util.Arrays; + +public class Sha1HashId { + private final byte[] sha1Hash; + + private final String hex; + + public Sha1HashId(byte[] sha1Hash) { + this.sha1Hash = sha1Hash; + hex = Hex.encodeHexString(sha1Hash); + } + + @Override + public String toString() { + return "Sha1HashId{" + "sha1Hash=" + hex + "}"; + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (!(o instanceof Sha1HashId)) + return false; + + Sha1HashId that = (Sha1HashId) o; + + if (!Arrays.equals(sha1Hash, that.sha1Hash)) + return false; + + return true; + } + + @Override + public int hashCode() { + return Arrays.hashCode(sha1Hash); + } + + public byte[] getSha1Hash() { + return sha1Hash; + } + + public ByteBuffer getSha1HashByteBuffer() { + return ByteBuffer.wrap(sha1Hash); + } + + public String toHexString() { + return hex; + } +} diff --git a/java/src/main/java/monasca/persister/repository/cassandra/CassandraAlarmRepo.java b/java/src/main/java/monasca/persister/repository/cassandra/CassandraAlarmRepo.java new file mode 100644 index 000000000000..96323099e13e --- /dev/null +++ b/java/src/main/java/monasca/persister/repository/cassandra/CassandraAlarmRepo.java @@ -0,0 +1,113 @@ +/* + * Copyright (c) 2017 SUSE LLC + * + * 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 monasca.persister.repository.cassandra; + +import java.security.NoSuchAlgorithmException; +import java.sql.SQLException; +import java.sql.Timestamp; + +import javax.inject.Inject; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.PropertyNamingStrategy; + +import io.dropwizard.setup.Environment; +import monasca.common.model.event.AlarmStateTransitionedEvent; +import monasca.persister.configuration.PersisterConfig; +import monasca.persister.repository.Repo; +import monasca.persister.repository.RepoException; + +/** + * This class is not thread safe. + * + */ +public class CassandraAlarmRepo extends CassandraRepo implements Repo<AlarmStateTransitionedEvent> { + + private static final Logger logger = LoggerFactory.getLogger(CassandraAlarmRepo.class); + + private static String EMPTY_REASON_DATA = "{}"; + + private static final int MAX_BYTES_PER_CHAR = 4; + private static final int MAX_LENGTH_VARCHAR = 65000; + + private int retention; + + private ObjectMapper objectMapper = new ObjectMapper(); + + @Inject + public CassandraAlarmRepo(CassandraCluster cluster, PersisterConfig config, Environment environment) + throws NoSuchAlgorithmException, SQLException { + super(cluster, environment, config.getCassandraDbConfiguration().getMaxWriteRetries(), + config.getAlarmHistoryConfiguration().getBatchSize()); + + this.retention = config.getCassandraDbConfiguration().getRetentionPolicy() * 24 * 3600; + + logger.debug("Instantiating " + this.getClass().getName()); + + this.objectMapper + .setPropertyNamingStrategy(PropertyNamingStrategy.CAMEL_CASE_TO_LOWER_CASE_WITH_UNDERSCORES); + + session = cluster.getAlarmsSession(); + + logger.debug(this.getClass().getName() + " is fully instantiated"); + + } + + public void addToBatch(AlarmStateTransitionedEvent message, String id) { + + String metricsString = getSerializedString(message.metrics, id); + + // Validate metricsString does not exceed a sufficient maximum upper bound + if (metricsString.length() * MAX_BYTES_PER_CHAR >= MAX_LENGTH_VARCHAR) { + metricsString = "[]"; + logger.warn("length of metricsString for alarm ID {} exceeds max length of {}", message.alarmId, + MAX_LENGTH_VARCHAR); + } + + String subAlarmsString = getSerializedString(message.subAlarms, id); + + if (subAlarmsString.length() * MAX_BYTES_PER_CHAR >= MAX_LENGTH_VARCHAR) { + subAlarmsString = "[]"; + logger.warn("length of subAlarmsString for alarm ID {} exceeds max length of {}", message.alarmId, + MAX_LENGTH_VARCHAR); + } + + queue.offerLast(cluster.getAlarmHistoryInsertStmt().bind(retention, metricsString, message.oldState.name(), + message.newState.name(), subAlarmsString, message.stateChangeReason, EMPTY_REASON_DATA, + message.tenantId, message.alarmId, new Timestamp(message.timestamp))); + } + + private String getSerializedString(Object o, String id) { + + try { + return this.objectMapper.writeValueAsString(o); + } catch (JsonProcessingException e) { + logger.error("[[}]: failed to serialize object {}", id, o, e); + return ""; + } + } + + @Override + public int flush(String id) throws RepoException { + return handleFlush(id); + } +} diff --git a/java/src/main/java/monasca/persister/repository/cassandra/CassandraCluster.java b/java/src/main/java/monasca/persister/repository/cassandra/CassandraCluster.java new file mode 100644 index 000000000000..ac20b098db63 --- /dev/null +++ b/java/src/main/java/monasca/persister/repository/cassandra/CassandraCluster.java @@ -0,0 +1,427 @@ +/* + * Copyright (c) 2017 SUSE LLC + * + * 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 monasca.persister.repository.cassandra; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.datastax.driver.core.BoundStatement; +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.Cluster.Builder; +import com.datastax.driver.core.CodecRegistry; +import com.datastax.driver.core.ConsistencyLevel; +import com.datastax.driver.core.HostDistance; +import com.datastax.driver.core.Metadata; +import com.datastax.driver.core.PlainTextAuthProvider; +import com.datastax.driver.core.PoolingOptions; +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.ProtocolOptions; +import com.datastax.driver.core.QueryOptions; +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.ResultSetFuture; +import com.datastax.driver.core.Row; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.SocketOptions; +import com.datastax.driver.core.TokenRange; +import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy; +import com.datastax.driver.core.policies.TokenAwarePolicy; +import com.datastax.driver.core.utils.Bytes; +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; +import com.google.common.collect.Lists; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; +import com.google.inject.Inject; + +import monasca.common.configuration.CassandraDbConfiguration; +import monasca.persister.configuration.PersisterConfig; + +public class CassandraCluster { + + private static final Logger logger = LoggerFactory.getLogger(CassandraCluster.class); + + private static final String MEASUREMENT_INSERT_CQL = "update monasca.measurements USING TTL ? " + + "set value = ?, value_meta = ?, region = ?, tenant_id = ?, metric_name = ?, dimensions = ? " + + "where metric_id = ? and time_stamp = ?"; + + private static final String MEASUREMENT_UPDATE_CQL = "update monasca.measurements USING TTL ? " + + "set value = ?, value_meta = ? " + "where metric_id = ? and time_stamp = ?"; + + private static final String METRICS_INSERT_CQL = "update monasca.metrics USING TTL ? " + + "set metric_id = ?, created_at = ?, updated_at = ? " + + "where region = ? and tenant_id = ? and metric_name = ? and dimensions = ? and dimension_names = ?"; + + private static final String METRICS_UPDATE_CQL = "update monasca.metrics USING TTL ? " + + "set updated_at = ? " + + "where region = ? and tenant_id = ? and metric_name = ? and dimensions = ? and dimension_names = ?"; + + private static final String DIMENSION_INSERT_CQL = "insert into monasca.dimensions " + + "(region, tenant_id, name, value) values (?, ?, ?, ?)"; + + private static final String DIMENSION_METRIC_INSERT_CQL = "insert into monasca.dimensions_metrics " + + " (region, tenant_id, dimension_name, dimension_value, metric_name) values (?, ?, ?, ?, ?)"; + + private static final String METRIC_DIMENSION_INSERT_CQL = "insert into monasca.metrics_dimensions " + + " (region, tenant_id, metric_name, dimension_name, dimension_value) values (?, ?, ?, ?, ?)"; + + private static final String INSERT_ALARM_STATE_HISTORY_SQL = "update monasca.alarm_state_history USING TTL ? " + + " set metric = ?, old_state = ?, new_state = ?, sub_alarms = ?, reason = ?, reason_data = ?" + + " where tenant_id = ? and alarm_id = ? and time_stamp = ?"; + + private static final String RETRIEVE_METRIC_DIMENSION_CQL = "select region, tenant_id, metric_name, " + + "dimension_name, dimension_value from metrics_dimensions " + + "WHERE token(region, tenant_id, metric_name) > ? and token(region, tenant_id, metric_name) <= ? "; + + private static final String RETRIEVE_METRIC_ID_CQL = "select distinct metric_id from measurements WHERE token(metric_id) > ? and token(metric_id) <= ?"; + + private static final String RETRIEVE_DIMENSION_CQL = "select region, tenant_id, name, value from dimensions"; + + private static final String NAME = "name"; + private static final String VALUE = "value"; + private static final String METRIC_ID = "metric_id"; + private static final String TENANT_ID_COLUMN = "tenant_id"; + private static final String METRIC_NAME = "metric_name"; + private static final String DIMENSION_NAME = "dimension_name"; + private static final String DIMENSION_VALUE = "dimension_value"; + private static final String REGION = "region"; + + private CassandraDbConfiguration dbConfig; + private Cluster cluster; + private Session metricsSession; + private Session alarmsSession; + + private TokenAwarePolicy lbPolicy; + + private PreparedStatement measurementInsertStmt; + private PreparedStatement measurementUpdateStmt; + private PreparedStatement metricInsertStmt; + private PreparedStatement metricUpdateStmt; + private PreparedStatement dimensionStmt; + private PreparedStatement dimensionMetricStmt; + private PreparedStatement metricDimensionStmt; + + private PreparedStatement retrieveMetricDimensionStmt; + private PreparedStatement retrieveMetricIdStmt; + + private PreparedStatement alarmHistoryInsertStmt; + + public Cache<String, Boolean> getMetricIdCache() { + return metricIdCache; + } + + public Cache<String, Boolean> getDimensionCache() { + return dimensionCache; + } + + public Cache<String, Boolean> getMetricDimensionCache() { + return metricDimensionCache; + } + + private final Cache<String, Boolean> metricIdCache; + + private final Cache<String, Boolean> dimensionCache; + + private final Cache<String, Boolean> metricDimensionCache; + + @Inject + public CassandraCluster(final PersisterConfig config) { + + this.dbConfig = config.getCassandraDbConfiguration(); + + QueryOptions qo = new QueryOptions(); + qo.setConsistencyLevel(ConsistencyLevel.valueOf(dbConfig.getConsistencyLevel())); + qo.setDefaultIdempotence(true); + + String[] contactPoints = dbConfig.getContactPoints(); + int retries = dbConfig.getMaxWriteRetries(); + Builder builder = Cluster.builder().addContactPoints(contactPoints).withPort(dbConfig.getPort()); + builder + .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(dbConfig.getConnectionTimeout()) + .setReadTimeoutMillis(dbConfig.getReadTimeout())); + builder.withQueryOptions(qo).withRetryPolicy(new MonascaRetryPolicy(retries, retries, retries)); + + lbPolicy = new TokenAwarePolicy( + DCAwareRoundRobinPolicy.builder().withLocalDc(dbConfig.getLocalDataCenter()).build()); + builder.withLoadBalancingPolicy(lbPolicy); + + String user = dbConfig.getUser(); + if (user != null && !user.isEmpty()) { + builder.withAuthProvider(new PlainTextAuthProvider(dbConfig.getUser(), dbConfig.getPassword())); + } + cluster = builder.build(); + + PoolingOptions poolingOptions = cluster.getConfiguration().getPoolingOptions(); + + poolingOptions.setConnectionsPerHost(HostDistance.LOCAL, dbConfig.getMaxConnections(), + dbConfig.getMaxConnections()).setConnectionsPerHost(HostDistance.REMOTE, + dbConfig.getMaxConnections(), dbConfig.getMaxConnections()); + + poolingOptions.setMaxRequestsPerConnection(HostDistance.LOCAL, dbConfig.getMaxRequests()) + .setMaxRequestsPerConnection(HostDistance.REMOTE, dbConfig.getMaxRequests()); + + metricsSession = cluster.connect(dbConfig.getKeySpace()); + + measurementInsertStmt = metricsSession.prepare(MEASUREMENT_INSERT_CQL).setIdempotent(true); + measurementUpdateStmt = metricsSession.prepare(MEASUREMENT_UPDATE_CQL).setIdempotent(true); + metricInsertStmt = metricsSession.prepare(METRICS_INSERT_CQL).setIdempotent(true); + metricUpdateStmt = metricsSession.prepare(METRICS_UPDATE_CQL).setIdempotent(true); + dimensionStmt = metricsSession.prepare(DIMENSION_INSERT_CQL).setIdempotent(true); + dimensionMetricStmt = metricsSession.prepare(DIMENSION_METRIC_INSERT_CQL).setIdempotent(true); + metricDimensionStmt = metricsSession.prepare(METRIC_DIMENSION_INSERT_CQL).setIdempotent(true); + + retrieveMetricIdStmt = metricsSession.prepare(RETRIEVE_METRIC_ID_CQL).setIdempotent(true); + retrieveMetricDimensionStmt = metricsSession.prepare(RETRIEVE_METRIC_DIMENSION_CQL) + .setIdempotent(true); + + alarmsSession = cluster.connect(dbConfig.getKeySpace()); + + alarmHistoryInsertStmt = alarmsSession.prepare(INSERT_ALARM_STATE_HISTORY_SQL).setIdempotent(true); + + metricIdCache = CacheBuilder.newBuilder() + .maximumSize(config.getCassandraDbConfiguration().getDefinitionMaxCacheSize()).build(); + + dimensionCache = CacheBuilder.newBuilder() + .maximumSize(config.getCassandraDbConfiguration().getDefinitionMaxCacheSize()).build(); + + metricDimensionCache = CacheBuilder.newBuilder() + .maximumSize(config.getCassandraDbConfiguration().getDefinitionMaxCacheSize()).build(); + + logger.info("loading cached definitions from db"); + + ExecutorService executor = Executors.newFixedThreadPool(250); + + //a majority of the ids are for metrics not actively receiving msgs anymore + //loadMetricIdCache(executor); + + loadDimensionCache(); + + loadMetricDimensionCache(executor); + + executor.shutdown(); + } + + public Session getMetricsSession() { + return metricsSession; + } + + public Session getAlarmsSession() { + return alarmsSession; + } + + public PreparedStatement getMeasurementInsertStmt() { + return measurementInsertStmt; + } + + public PreparedStatement getMeasurementUpdateStmt() { + return measurementUpdateStmt; + } + + public PreparedStatement getMetricInsertStmt() { + return metricInsertStmt; + } + + public PreparedStatement getMetricUpdateStmt() { + return metricUpdateStmt; + } + + public PreparedStatement getDimensionStmt() { + return dimensionStmt; + } + + public PreparedStatement getDimensionMetricStmt() { + return dimensionMetricStmt; + } + + public PreparedStatement getMetricDimensionStmt() { + return metricDimensionStmt; + } + + public PreparedStatement getAlarmHistoryInsertStmt() { + return alarmHistoryInsertStmt; + } + + public ProtocolOptions getProtocolOptions() { + return cluster.getConfiguration().getProtocolOptions(); + } + + public CodecRegistry getCodecRegistry() { + return cluster.getConfiguration().getCodecRegistry(); + } + + public Metadata getMetaData() { + return cluster.getMetadata(); + } + + public TokenAwarePolicy getLoadBalancePolicy() { + return lbPolicy; + } + + private void loadMetricIdCache(ExecutorService executor) { + final AtomicInteger tasks = new AtomicInteger(0); + logger.info("Found token ranges: " + cluster.getMetadata().getTokenRanges().size()); + for (TokenRange range : cluster.getMetadata().getTokenRanges()) { + List<BoundStatement> queries = rangeQuery(retrieveMetricIdStmt, range); + for (BoundStatement query : queries) { + tasks.incrementAndGet(); + logger.info("adding a metric id reading task, total: " + tasks.get()); + + ResultSetFuture future = metricsSession.executeAsync(query); + + Futures.addCallback(future, new FutureCallback<ResultSet>() { + @Override + public void onSuccess(ResultSet result) { + for (Row row : result) { + String id = Bytes.toHexString(row.getBytes(METRIC_ID)); + if (id != null) { + //remove '0x' + metricIdCache.put(id.substring(2), Boolean.TRUE); + } + } + + tasks.decrementAndGet(); + + logger.info("completed a metric id read task. Remaining tasks: " + tasks.get()); + } + + @Override + public void onFailure(Throwable t) { + logger.error("Failed to execute query to load metric id cache.", t); + + tasks.decrementAndGet(); + + logger.info("Failed a metric id read task. Remaining tasks: " + tasks.get()); + } + }, executor); + + } + } + + while (tasks.get() > 0) { + logger.debug("waiting for more metric id load tasks: " + tasks.get()); + + try { + Thread.sleep(3000); + } catch (InterruptedException e) { + logger.warn("load metric cache was interrupted", e); + } + } + + logger.info("loaded metric id cache from database: " + metricIdCache.size()); + } + + private List<BoundStatement> rangeQuery(PreparedStatement rangeStmt, TokenRange range) { + List<BoundStatement> res = Lists.newArrayList(); + for (TokenRange subRange : range.unwrap()) { + res.add(rangeStmt.bind(subRange.getStart(), subRange.getEnd())); + } + return res; + } + + private void loadDimensionCache() { + + ResultSet results = metricsSession.execute(RETRIEVE_DIMENSION_CQL); + + for (Row row : results) { + String key = getDimnesionEntryKey(row.getString(REGION), row.getString(TENANT_ID_COLUMN), + row.getString(NAME), row.getString(VALUE)); + dimensionCache.put(key, Boolean.TRUE); + } + + logger.info("loaded dimension cache from database: " + dimensionCache.size()); + } + + public String getDimnesionEntryKey(String region, String tenantId, String name, String value) { + StringBuilder sb = new StringBuilder(); + sb.append(region).append('\0'); + sb.append(tenantId).append('\0'); + sb.append(name).append('\0'); + sb.append(value); + return sb.toString(); + } + + private void loadMetricDimensionCache(ExecutorService executor) { + + final AtomicInteger tasks = new AtomicInteger(0); + + for (TokenRange range : cluster.getMetadata().getTokenRanges()) { + List<BoundStatement> queries = rangeQuery(retrieveMetricDimensionStmt, range); + for (BoundStatement query : queries) { + tasks.incrementAndGet(); + + logger.info("Adding a metric dimnesion read task, total: " + tasks.get()); + + ResultSetFuture future = metricsSession.executeAsync(query); + + Futures.addCallback(future, new FutureCallback<ResultSet>() { + @Override + public void onSuccess(ResultSet result) { + for (Row row : result) { + String key = getMetricDimnesionEntryKey(row.getString(REGION), + row.getString(TENANT_ID_COLUMN), row.getString(METRIC_NAME), + row.getString(DIMENSION_NAME), row.getString(DIMENSION_VALUE)); + metricDimensionCache.put(key, Boolean.TRUE); + } + + tasks.decrementAndGet(); + + logger.info("Completed a metric dimension read task. Remaining tasks: " + tasks.get()); + } + + @Override + public void onFailure(Throwable t) { + logger.error("Failed to execute query to load metric id cache.", t); + + tasks.decrementAndGet(); + + logger.info("Failed a metric dimension read task. Remaining tasks: " + tasks.get()); + } + }, executor); + + } + } + + while (tasks.get() > 0) { + + logger.debug("waiting for metric dimension cache to load ..."); + + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + logger.warn("load metric dimension cache was interrupted", e); + } + } + + logger.info("loaded metric dimension cache from database: " + metricDimensionCache.size()); + } + + public String getMetricDimnesionEntryKey(String region, String tenantId, String metricName, + String dimensionName, String dimensionValue) { + StringBuilder sb = new StringBuilder(); + sb.append(region).append('\0'); + sb.append(tenantId).append('\0'); + sb.append(metricName).append('\0'); + sb.append(dimensionName).append('\0'); + sb.append(dimensionValue); + return sb.toString(); + } +} diff --git a/java/src/main/java/monasca/persister/repository/cassandra/CassandraMetricBatch.java b/java/src/main/java/monasca/persister/repository/cassandra/CassandraMetricBatch.java new file mode 100644 index 000000000000..4a275676cacf --- /dev/null +++ b/java/src/main/java/monasca/persister/repository/cassandra/CassandraMetricBatch.java @@ -0,0 +1,207 @@ +/* + * Copyright (c) 2017 SUSE LLC + * + * 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 monasca.persister.repository.cassandra; + +import java.nio.ByteBuffer; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.datastax.driver.core.BatchStatement; +import com.datastax.driver.core.BatchStatement.Type; +import com.datastax.driver.core.BoundStatement; +import com.datastax.driver.core.CodecRegistry; +import com.datastax.driver.core.Host; +import com.datastax.driver.core.Metadata; +import com.datastax.driver.core.ProtocolOptions; +import com.datastax.driver.core.Token; +import com.datastax.driver.core.policies.TokenAwarePolicy; + +public class CassandraMetricBatch { + private static Logger logger = LoggerFactory.getLogger(CassandraMetricBatch.class); + + ProtocolOptions protocol; + CodecRegistry codec; + Metadata metadata; + TokenAwarePolicy policy; + int batchLimit; + + Map<Token, Deque<BatchStatement>> metricQueries; + Map<Token, Deque<BatchStatement>> dimensionQueries; + Map<Token, Deque<BatchStatement>> dimensionMetricQueries; + Map<Token, Deque<BatchStatement>> metricDimensionQueries; + Map<Set<Host>, Deque<BatchStatement>> measurementQueries; + + public CassandraMetricBatch(Metadata metadata, ProtocolOptions protocol, CodecRegistry codec, + TokenAwarePolicy lbPolicy, int batchLimit) { + this.protocol = protocol; + this.codec = codec; + this.metadata = metadata; + this.policy = lbPolicy; + metricQueries = new HashMap<>(); + this.batchLimit = batchLimit; + + metricQueries = new HashMap<>(); + dimensionQueries = new HashMap<>(); + dimensionMetricQueries = new HashMap<>(); + metricDimensionQueries = new HashMap<>(); + measurementQueries = new HashMap<>(); + } + + public void addMetricQuery(BoundStatement s) { + batchQueryByToken(s, metricQueries); + } + + public void addDimensionQuery(BoundStatement s) { + batchQueryByToken(s, dimensionQueries); + } + + public void addDimensionMetricQuery(BoundStatement s) { + batchQueryByToken(s, dimensionMetricQueries); + } + + public void addMetricDimensionQuery(BoundStatement s) { + batchQueryByToken(s, metricDimensionQueries); + } + + public void addMeasurementQuery(BoundStatement s) { + batchQueryByReplica(s, measurementQueries); + } + + private void batchQueryByToken(BoundStatement s, Map<Token, Deque<BatchStatement>> batchedQueries) { + ByteBuffer b = s.getRoutingKey(protocol.getProtocolVersion(), codec); + Token token = metadata.newToken(b); + Deque<BatchStatement> queue = batchedQueries.get(token); + if (queue == null) { + queue = new ArrayDeque<BatchStatement>(); + BatchStatement bs = new BatchStatement(Type.UNLOGGED); + bs.add(s); + queue.offer(bs); + batchedQueries.put(token, queue); + } else { + BatchStatement bs = queue.getLast(); + if (bs.size() < batchLimit) { + bs.add(s); + } else { + bs = new BatchStatement(Type.UNLOGGED); + bs.add(s); + queue.offerLast(bs); + } + } + } + + private void batchQueryByReplica(BoundStatement s, + Map<Set<Host>, Deque<BatchStatement>> batchedQueries) { + Iterator<Host> it = policy.newQueryPlan(s.getKeyspace(), s); + Set<Host> hosts = new HashSet<>(); + + while (it.hasNext()) { + hosts.add(it.next()); + } + + Deque<BatchStatement> queue = batchedQueries.get(hosts); + if (queue == null) { + queue = new ArrayDeque<BatchStatement>(); + BatchStatement bs = new BatchStatement(Type.UNLOGGED); + bs.add(s); + queue.offer(bs); + batchedQueries.put(hosts, queue); + } else { + BatchStatement bs = queue.getLast(); + if (bs.size() < 30) { + bs.add(s); + } else { + bs = new BatchStatement(Type.UNLOGGED); + bs.add(s); + queue.offerLast(bs); + } + } + } + + public void clear() { + metricQueries.clear(); + dimensionQueries.clear(); + dimensionMetricQueries.clear(); + metricDimensionQueries.clear(); + measurementQueries.clear(); + } + + public List<Deque<BatchStatement>> getAllBatches() { + logTokenBatchMap("metric batches", metricQueries); + logTokenBatchMap("dimension batches", dimensionQueries); + logTokenBatchMap("dimension metric batches", dimensionMetricQueries); + logTokenBatchMap("metric dimension batches", metricDimensionQueries); + logReplicaBatchMap("measurement batches", measurementQueries); + + ArrayList<Deque<BatchStatement>> list = new ArrayList<>(); + list.addAll(metricQueries.values()); + list.addAll(dimensionQueries.values()); + list.addAll(dimensionMetricQueries.values()); + list.addAll(metricDimensionQueries.values()); + list.addAll(measurementQueries.values()); + return list; + } + + private void logTokenBatchMap(String name, Map<Token, Deque<BatchStatement>> map) { + if (logger.isDebugEnabled()) { + StringBuilder sb = new StringBuilder(name); + sb.append(": Size: ").append(map.size()); + sb.append("; Tokens: |"); + for (Entry<Token, Deque<BatchStatement>> entry : map.entrySet()) { + sb.append(entry.getKey().toString()).append(":"); + for (BatchStatement bs : entry.getValue()) { + sb.append(bs.size()).append(","); + } + sb.append("|."); + } + + logger.debug(sb.toString()); + } + } + + private void logReplicaBatchMap(String name, Map<Set<Host>, Deque<BatchStatement>> map) { + if (logger.isDebugEnabled()) { + StringBuilder sb = new StringBuilder(name); + sb.append(": Size: ").append(map.size()); + sb.append(". Replicas: |"); + for (Entry<Set<Host>, Deque<BatchStatement>> entry : map.entrySet()) { + for (Host host : entry.getKey()) { + sb.append(host.getAddress().toString()).append(","); + } + sb.append(":"); + for (BatchStatement bs : entry.getValue()) { + sb.append(bs.size()).append(","); + } + + sb.append("|"); + + } + logger.debug(sb.toString()); + } + } +} diff --git a/java/src/main/java/monasca/persister/repository/cassandra/CassandraMetricRepo.java b/java/src/main/java/monasca/persister/repository/cassandra/CassandraMetricRepo.java new file mode 100644 index 000000000000..d096e78448b0 --- /dev/null +++ b/java/src/main/java/monasca/persister/repository/cassandra/CassandraMetricRepo.java @@ -0,0 +1,336 @@ +/* + * Copyright (c) 2017 SUSE LLC + * + * 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 monasca.persister.repository.cassandra; + +import java.security.NoSuchAlgorithmException; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.util.ArrayList; +import java.util.Deque; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.TreeMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import javax.inject.Inject; + +import org.apache.commons.codec.digest.DigestUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.codahale.metrics.Meter; +import com.datastax.driver.core.BatchStatement; +import com.datastax.driver.core.BoundStatement; +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.ResultSetFuture; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; + +import io.dropwizard.setup.Environment; +import monasca.common.model.metric.Metric; +import monasca.common.model.metric.MetricEnvelope; +import monasca.persister.configuration.PersisterConfig; +import monasca.persister.repository.Repo; +import monasca.persister.repository.RepoException; +import monasca.persister.repository.Sha1HashId; + +public class CassandraMetricRepo extends CassandraRepo implements Repo<MetricEnvelope> { + + private static final Logger logger = LoggerFactory.getLogger(CassandraMetricRepo.class); + + public static final int MAX_COLUMN_LENGTH = 255; + public static final int MAX_VALUE_META_LENGTH = 2048; + + private static final String TENANT_ID = "tenantId"; + private static final String REGION = "region"; + private static final String EMPTY_STR = ""; + + private int retention; + + private CassandraMetricBatch batches; + + private int metricCount; + + private final ObjectMapper objectMapper = new ObjectMapper(); + + public final Meter measurementMeter; + public final Meter metricCacheMissMeter; + public final Meter metricCacheHitMeter; + public final Meter dimensionCacheMissMeter; + public final Meter dimensionCacheHitMeter; + public final Meter metricDimensionCacheMissMeter; + public final Meter metricDimensionCacheHitMeter; + + @Inject + public CassandraMetricRepo(CassandraCluster cluster, PersisterConfig config, Environment environment) + throws NoSuchAlgorithmException, SQLException { + + super(cluster, environment, config.getCassandraDbConfiguration().getMaxWriteRetries(), + config.getMetricConfiguration().getBatchSize()); + + logger.debug("Instantiating " + this.getClass().getName()); + + this.retention = config.getCassandraDbConfiguration().getRetentionPolicy() * 24 * 3600; + + this.measurementMeter = this.environment.metrics() + .meter(this.getClass().getName() + "." + "measurement-meter"); + + this.metricCacheMissMeter = this.environment.metrics() + .meter(this.getClass().getName() + "." + "definition-cache-miss-meter"); + + this.metricCacheHitMeter = this.environment.metrics() + .meter(this.getClass().getName() + "." + "definition-cache-hit-meter"); + + this.dimensionCacheMissMeter = this.environment.metrics() + .meter(this.getClass().getName() + "." + "dimension-cache-miss-meter"); + + this.dimensionCacheHitMeter = this.environment.metrics() + .meter(this.getClass().getName() + "." + "dimension-cache-hit-meter"); + + this.metricDimensionCacheMissMeter = this.environment.metrics() + .meter(this.getClass().getName() + "." + "metric-dimension-cache-miss-meter"); + + this.metricDimensionCacheHitMeter = this.environment.metrics() + .meter(this.getClass().getName() + "." + "metric-dimension-cache-hit-meter"); + + session = cluster.getMetricsSession(); + + metricCount = 0; + + batches = new CassandraMetricBatch(cluster.getMetaData(), cluster.getProtocolOptions(), + cluster.getCodecRegistry(), cluster.getLoadBalancePolicy(), + config.getCassandraDbConfiguration().getMaxBatches()); + + + + logger.debug(this.getClass().getName() + " is fully instantiated"); + } + + @Override + public void addToBatch(MetricEnvelope metricEnvelope, String id) { + Metric metric = metricEnvelope.metric; + Map<String, Object> metaMap = metricEnvelope.meta; + + String tenantId = getMeta(TENANT_ID, metric, metaMap, id); + String region = getMeta(REGION, metric, metaMap, id); + String metricName = metric.getName(); + TreeMap<String, String> dimensions = metric.getDimensions() == null ? new TreeMap<String, String>() + : new TreeMap<>(metric.getDimensions()); + + StringBuilder sb = new StringBuilder(region).append(tenantId).append(metricName); + + Iterator<String> it = dimensions.keySet().iterator(); + while (it.hasNext()) { + String k = it.next(); + sb.append(k).append(dimensions.get(k)); + } + + byte[] defIdSha = DigestUtils.sha(sb.toString()); + Sha1HashId defIdShaHash = new Sha1HashId(defIdSha); + + if (cluster.getMetricIdCache().getIfPresent(defIdShaHash.toHexString()) == null) { + addDefinitionToBatch(defIdShaHash, metricName, dimensions, tenantId, region, id, + metric.getTimestamp()); + batches.addMeasurementQuery(buildMeasurementInsertQuery(defIdShaHash, metric.getTimestamp(), + metric.getValue(), metric.getValueMeta(), region, tenantId, metricName, dimensions, id)); + } else { + metricCacheHitMeter.mark(); + batches.addMetricQuery(cluster.getMetricUpdateStmt().bind(retention, + new Timestamp(metric.getTimestamp()), region, tenantId, metricName, + getDimensionList(dimensions), new ArrayList<>(dimensions.keySet()))); + batches.addMeasurementQuery(buildMeasurementUpdateQuery(defIdShaHash, metric.getTimestamp(), + metric.getValue(), metric.getValueMeta(), id)); + } + + metricCount++; + } + + private String getMeta(String name, Metric metric, Map<String, Object> meta, String id) { + if (meta.containsKey(name)) { + return (String) meta.get(name); + } else { + logger.warn( + "[{}]: failed to find {} in message envelope meta data. metric message may be malformed. " + + "setting {} to empty string.", + id, name); + logger.warn("[{}]: metric: {}", id, metric.toString()); + logger.warn("[{}]: meta: {}", id, meta.toString()); + return EMPTY_STR; + } + } + + private BoundStatement buildMeasurementUpdateQuery(Sha1HashId defId, long timeStamp, double value, + Map<String, String> valueMeta, String id) { + + String valueMetaString = getValueMetaString(valueMeta, id); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: adding metric to batch: metric id: {}, time: {}, value: {}, value meta {}", + id, defId.toHexString(), timeStamp, value, valueMetaString); + } + + return cluster.getMeasurementUpdateStmt().bind(retention, value, valueMetaString, + defId.getSha1HashByteBuffer(), new Timestamp(timeStamp)); + } + + private BoundStatement buildMeasurementInsertQuery(Sha1HashId defId, long timeStamp, double value, + Map<String, String> valueMeta, String region, String tenantId, String metricName, + Map<String, String> dimensions, String id) { + + String valueMetaString = getValueMetaString(valueMeta, id); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: adding metric to batch: metric id: {}, time: {}, value: {}, value meta {}", + id, defId.toHexString(), timeStamp, value, valueMetaString); + } + + measurementMeter.mark(); + return cluster.getMeasurementInsertStmt().bind(retention, value, valueMetaString, region, tenantId, + metricName, getDimensionList(dimensions), defId.getSha1HashByteBuffer(), + new Timestamp(timeStamp)); + } + + private String getValueMetaString(Map<String, String> valueMeta, String id) { + + String valueMetaString = ""; + + if (valueMeta != null && !valueMeta.isEmpty()) { + + try { + + valueMetaString = this.objectMapper.writeValueAsString(valueMeta); + if (valueMetaString.length() > MAX_VALUE_META_LENGTH) { + logger.error("[{}]: Value meta length {} longer than maximum {}, dropping value meta", id, + valueMetaString.length(), MAX_VALUE_META_LENGTH); + return ""; + } + + } catch (JsonProcessingException e) { + + logger.error("[{}]: Failed to serialize value meta {}, dropping value meta from measurement", + id, valueMeta); + } + } + + return valueMetaString; + } + + private void addDefinitionToBatch(Sha1HashId defId, String metricName, Map<String, String> dimensions, + String tenantId, String region, String id, long timestamp) { + + metricCacheMissMeter.mark(); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: adding definition to batch: defId: {}, name: {}, tenantId: {}, region: {}", + id, defId.toHexString(), metricName, tenantId, region); + } + + Timestamp ts = new Timestamp(timestamp); + batches.addMetricQuery( + cluster.getMetricInsertStmt().bind(retention, defId.getSha1HashByteBuffer(), ts, ts, region, + tenantId, metricName, getDimensionList(dimensions), new ArrayList<>(dimensions.keySet()))); + + for (Map.Entry<String, String> entry : dimensions.entrySet()) { + String name = entry.getKey(); + String value = entry.getValue(); + + String dimensionKey = cluster.getDimnesionEntryKey(region, tenantId, name, value); + if (cluster.getDimensionCache().getIfPresent(dimensionKey) != null) { + dimensionCacheHitMeter.mark(); + + } else { + dimensionCacheMissMeter.mark(); + if (logger.isDebugEnabled()) { + logger.debug("[{}]: adding dimension to batch: defId: {}, name: {}, value: {}", id, + defId.toHexString(), name, value); + } + batches.addDimensionQuery(cluster.getDimensionStmt().bind(region, tenantId, name, value)); + cluster.getDimensionCache().put(dimensionKey, Boolean.TRUE); + } + + String metricDimensionKey = cluster.getMetricDimnesionEntryKey(region, tenantId, metricName, name, value); + if (cluster.getMetricDimensionCache().getIfPresent(metricDimensionKey) != null) { + metricDimensionCacheHitMeter.mark(); + } else { + metricDimensionCacheMissMeter.mark(); + batches.addDimensionMetricQuery( + cluster.getDimensionMetricStmt().bind(region, tenantId, name, value, metricName)); + + batches.addMetricDimensionQuery( + cluster.getMetricDimensionStmt().bind(region, tenantId, metricName, name, value)); + cluster.getMetricDimensionCache().put(metricDimensionKey, Boolean.TRUE); + } + } + + String metricId = defId.toHexString(); + cluster.getMetricIdCache().put(metricId, Boolean.TRUE); + } + + public List<String> getDimensionList(Map<String, String> dimensions) { + List<String> list = new ArrayList<>(dimensions.size()); + for (Entry<String, String> dim : dimensions.entrySet()) { + list.add(new StringBuffer(dim.getKey()).append('\t').append(dim.getValue()).toString()); + } + return list; + } + + @Override + public int flush(String id) throws RepoException { + long startTime = System.nanoTime(); + List<ResultSetFuture> results = new ArrayList<>(); + List<Deque<BatchStatement>> list = batches.getAllBatches(); + for (Deque<BatchStatement> q : list) { + BatchStatement b; + while ((b = q.poll()) != null) { + results.add(session.executeAsync(b)); + } + } + + List<ListenableFuture<ResultSet>> futures = Futures.inCompletionOrder(results); + + boolean cancel = false; + Exception ex = null; + for (ListenableFuture<ResultSet> future : futures) { + if (cancel) { + future.cancel(false); + continue; + } + try { + future.get(); + } catch (InterruptedException | ExecutionException e) { + cancel = true; + ex = e; + } + } + + this.commitTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + + if (ex != null) { + metricFailed.inc(metricCount); + throw new RepoException(ex); + } + + batches.clear(); + int flushCnt = metricCount; + metricCount = 0; + metricCompleted.inc(flushCnt); + return flushCnt; + } +} diff --git a/java/src/main/java/monasca/persister/repository/cassandra/CassandraRepo.java b/java/src/main/java/monasca/persister/repository/cassandra/CassandraRepo.java new file mode 100644 index 000000000000..e536d3e5f73a --- /dev/null +++ b/java/src/main/java/monasca/persister/repository/cassandra/CassandraRepo.java @@ -0,0 +1,210 @@ +/* + * Copyright (c) 2017 SUSE LLC + * + * 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 monasca.persister.repository.cassandra; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Timer; +import com.datastax.driver.core.BatchStatement; +import com.datastax.driver.core.BatchStatement.Type; +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.ResultSetFuture; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.Statement; +import com.datastax.driver.core.exceptions.BootstrappingException; +import com.datastax.driver.core.exceptions.DriverException; +import com.datastax.driver.core.exceptions.NoHostAvailableException; +import com.datastax.driver.core.exceptions.OperationTimedOutException; +import com.datastax.driver.core.exceptions.OverloadedException; +import com.datastax.driver.core.exceptions.QueryConsistencyException; +import com.datastax.driver.core.exceptions.UnavailableException; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; + +import io.dropwizard.setup.Environment; +import monasca.persister.repository.RepoException; + +public abstract class CassandraRepo { + private static Logger logger = LoggerFactory.getLogger(CassandraRepo.class); + + final Environment environment; + + final Timer commitTimer; + + CassandraCluster cluster; + Session session; + + int maxWriteRetries; + + int batchSize; + + long lastFlushTimeStamp; + + Deque<Statement> queue; + + Counter metricCompleted; + + Counter metricFailed; + + public CassandraRepo(CassandraCluster cluster, Environment env, int maxWriteRetries, int batchSize) { + this.cluster = cluster; + this.maxWriteRetries = maxWriteRetries; + this.batchSize = batchSize; + + this.environment = env; + + this.commitTimer = this.environment.metrics().timer(getClass().getName() + "." + "commit-timer"); + + lastFlushTimeStamp = System.currentTimeMillis(); + + queue = new ArrayDeque<>(batchSize); + + this.metricCompleted = environment.metrics() + .counter(getClass().getName() + "." + "metrics-persisted-counter"); + + this.metricFailed = environment.metrics() + .counter(getClass().getName() + "." + "metrics-failed-counter"); + } + + protected void executeQuery(String id, Statement query, long startTime) throws DriverException { + _executeQuery(id, query, startTime, 0); + } + + private void _executeQuery(final String id, final Statement query, final long startTime, + final int retryCount) { + try { + session.execute(query); + + commitTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + + // ResultSetFuture future = session.executeAsync(query); + + // Futures.addCallback(future, new FutureCallback<ResultSet>() { + // @Override + // public void onSuccess(ResultSet result) { + // metricCompleted.inc(); + // commitTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + // } + // + // @Override + // public void onFailure(Throwable t) { + // if (t instanceof NoHostAvailableException | t instanceof + // BootstrappingException + // | t instanceof OverloadedException | t instanceof QueryConsistencyException + // | t instanceof UnavailableException) { + // retryQuery(id, query, startTime, retryCount, (DriverException) t); + // } else { + // metricFailed.inc(); + // commitTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + // logger.error("Failed to execute query.", t); + // } + // } + // }, MoreExecutors.sameThreadExecutor()); + + } catch (NoHostAvailableException | BootstrappingException | OverloadedException + | QueryConsistencyException | UnavailableException | OperationTimedOutException e) { + retryQuery(id, query, startTime, retryCount, e); + } catch (DriverException e) { + metricFailed.inc(((BatchStatement) query).size()); + commitTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + throw e; + } + } + + private void retryQuery(String id, Statement query, final long startTime, int retryCount, + DriverException e) throws DriverException { + if (retryCount >= maxWriteRetries) { + logger.error("[{}]: Query aborted after {} retry: ", id, retryCount, e.getMessage()); + metricFailed.inc(((BatchStatement) query).size()); + commitTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + throw e; + } else { + logger.warn("[{}]: Query failed, retrying {} of {}: {} ", id, retryCount, maxWriteRetries, + e.getMessage()); + + try { + Thread.sleep(1000 * (1 << retryCount)); + } catch (InterruptedException ie) { + logger.debug("[{}]: Interrupted: {}", id, ie); + } + _executeQuery(id, query, startTime, retryCount++); + } + } + + public int handleFlush_batch(String id) { + Statement query; + int flushedCount = 0; + + BatchStatement batch = new BatchStatement(Type.UNLOGGED); + while ((query = queue.poll()) != null) { + flushedCount++; + batch.add(query); + } + + executeQuery(id, batch, System.nanoTime()); + + metricCompleted.inc(flushedCount); + + return flushedCount; + } + + public int handleFlush(String id) throws RepoException { + long startTime = System.nanoTime(); + + int flushedCount = 0; + List<ResultSetFuture> results = new ArrayList<>(queue.size()); + Statement query; + while ((query = queue.poll()) != null) { + flushedCount++; + results.add(session.executeAsync(query)); + } + + List<ListenableFuture<ResultSet>> futures = Futures.inCompletionOrder(results); + + boolean cancel = false; + Exception ex = null; + for (ListenableFuture<ResultSet> future : futures) { + if (cancel) { + future.cancel(false); + continue; + } + try { + future.get(); + } catch (InterruptedException | ExecutionException e) { + cancel = true; + ex = e; + } + } + + commitTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + + if (ex != null) { + throw new RepoException(ex); + } + return flushedCount; + } +} diff --git a/java/src/main/java/monasca/persister/repository/cassandra/MonascaRetryPolicy.java b/java/src/main/java/monasca/persister/repository/cassandra/MonascaRetryPolicy.java new file mode 100644 index 000000000000..f2d884aa25bb --- /dev/null +++ b/java/src/main/java/monasca/persister/repository/cassandra/MonascaRetryPolicy.java @@ -0,0 +1,77 @@ +package monasca.persister.repository.cassandra; + +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.ConsistencyLevel; +import com.datastax.driver.core.Statement; +import com.datastax.driver.core.WriteType; +import com.datastax.driver.core.exceptions.DriverException; +import com.datastax.driver.core.policies.RetryPolicy; + +public class MonascaRetryPolicy implements RetryPolicy { + + private final int readAttempts; + private final int writeAttempts; + private final int unavailableAttempts; + + public MonascaRetryPolicy(int readAttempts, int writeAttempts, int unavailableAttempts) { + super(); + this.readAttempts = readAttempts; + this.writeAttempts = writeAttempts; + this.unavailableAttempts = unavailableAttempts; + } + + @Override + public RetryDecision onReadTimeout(Statement stmnt, ConsistencyLevel cl, int requiredResponses, + int receivedResponses, boolean dataReceived, int rTime) { + if (dataReceived) { + return RetryDecision.ignore(); + } else if (rTime < readAttempts) { + return receivedResponses >= requiredResponses ? RetryDecision.retry(cl) + : RetryDecision.rethrow(); + } else { + return RetryDecision.rethrow(); + } + + } + + @Override + public RetryDecision onWriteTimeout(Statement stmnt, ConsistencyLevel cl, WriteType wt, + int requiredResponses, int receivedResponses, int wTime) { + if (wTime >= writeAttempts) + return RetryDecision.rethrow(); + + return RetryDecision.retry(cl); + } + + @Override + public RetryDecision onUnavailable(Statement stmnt, ConsistencyLevel cl, int requiredResponses, + int receivedResponses, int uTime) { + if (uTime == 0) { + return RetryDecision.tryNextHost(cl); + } else if (uTime <= unavailableAttempts) { + return RetryDecision.retry(cl); + } else { + return RetryDecision.rethrow(); + } + } + + /** + * {@inheritDoc} + */ + @Override + public RetryDecision onRequestError(Statement statement, ConsistencyLevel cl, DriverException e, + int nbRetry) { + return RetryDecision.tryNextHost(cl); + } + + @Override + public void init(Cluster cluster) { + // nothing to do + } + + @Override + public void close() { + // nothing to do + } + +} diff --git a/java/src/main/java/monasca/persister/repository/vertica/VerticaMetricRepo.java b/java/src/main/java/monasca/persister/repository/vertica/VerticaMetricRepo.java index 52baf7eecbdc..a0c22e4bf7aa 100644 --- a/java/src/main/java/monasca/persister/repository/vertica/VerticaMetricRepo.java +++ b/java/src/main/java/monasca/persister/repository/vertica/VerticaMetricRepo.java @@ -1,6 +1,8 @@ /* * (C) Copyright 2014-2016 Hewlett Packard Enterprise Development LP * + * (C) Copyright 2017 SUSE LLC. + * * 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 @@ -50,6 +52,7 @@ import monasca.common.model.metric.MetricEnvelope; import monasca.persister.configuration.PersisterConfig; import monasca.persister.repository.Repo; import monasca.persister.repository.RepoException; +import monasca.persister.repository.Sha1HashId; public class VerticaMetricRepo extends VerticaRepo implements Repo<MetricEnvelope> { diff --git a/monasca_persister/conf/cassandra.py b/monasca_persister/conf/cassandra.py index 93cdc325bc1e..09bbe18ae1dc 100644 --- a/monasca_persister/conf/cassandra.py +++ b/monasca_persister/conf/cassandra.py @@ -1,6 +1,7 @@ # (C) Copyright 2016 Hewlett Packard Enterprise Development Company LP # Copyright 2017 FUJITSU LIMITED -# +# (C) Copyright 2017 SUSE LLC + # 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 @@ -17,13 +18,47 @@ from oslo_config import cfg cassandra_opts = [ - cfg.ListOpt('cluster_ip_addresses', + cfg.ListOpt('contact_points', help='Comma separated list of Cassandra node IP addresses', default=['127.0.0.1'], item_type=cfg.IPOpt), + cfg.IntOpt('port', + help='Cassandra port number', + default=8086), cfg.StrOpt('keyspace', - help='keyspace where metric are stored', - default='monasca')] + help='Keyspace name where metrics are stored', + default='monasca'), + cfg.StrOpt('user', + help='Cassandra user name', + default=''), + cfg.StrOpt('password', + help='Cassandra password', + secret=True, + default=''), + cfg.IntOpt('connection_timeout', + help='Cassandra timeout in seconds when creating a new connection', + default=5), + cfg.IntOpt('read_timeout', + help='Cassandra read timeout in seconds', + default=60), + cfg.IntOpt('max_write_retries', + help='Maximum number of retries in write ops', + default=1), + cfg.IntOpt('max_definition_cache_size', + help='Maximum number of cached metric definition entries in memory', + default=20000000), + cfg.IntOpt('retention_policy', + help='Data retention period in days', + default=45), + cfg.StrOpt('consistency_level', + help='Cassandra default consistency level', + default='ONE'), + cfg.StrOpt('local_data_center', + help='Cassandra local data center name'), + cfg.IntOpt('max_batches', + help='Maximum batch size in Cassandra', + default=250), +] cassandra_group = cfg.OptGroup(name='cassandra') diff --git a/monasca_persister/conf/kafka_alarm_history.py b/monasca_persister/conf/kafka_alarm_history.py index 51ce9845f5a4..e5509e6a114d 100644 --- a/monasca_persister/conf/kafka_alarm_history.py +++ b/monasca_persister/conf/kafka_alarm_history.py @@ -1,5 +1,6 @@ # (C) Copyright 2016-2017 Hewlett Packard Enterprise Development LP # Copyright 2017 FUJITSU LIMITED +# (C) Copyright 2017 SUSE LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -37,7 +38,10 @@ kafka_alarm_history_opts = [ default='alarm-state-transitions'), cfg.StrOpt('zookeeper_path', help='Path in zookeeper for kafka consumer group partitioning algorithm', - default='/persister_partitions/$kafka_alarm_history.topic') + default='/persister_partitions/$kafka_alarm_history.topic'), + cfg.IntOpt('batch_size', + help='Maximum number of alarm state history messages to buffer before writing to database', + default=1), ] diff --git a/monasca_persister/conf/kafka_common.py b/monasca_persister/conf/kafka_common.py index 0c2cfc7620bc..534d59268a1b 100644 --- a/monasca_persister/conf/kafka_common.py +++ b/monasca_persister/conf/kafka_common.py @@ -1,5 +1,6 @@ # (C) Copyright 2016-2017 Hewlett Packard Enterprise Development LP # Copyright 2017 FUJITSU LIMITED +# (C) Copyright 2017 SUSE LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -26,9 +27,6 @@ kafka_common_opts = [ help='id of persister kafka client', advanced=True, default='monasca-persister'), - cfg.IntOpt('database_batch_size', - help='Maximum number of metric to buffer before writing to database', - default=1000), cfg.IntOpt('max_wait_time_seconds', help='Maximum wait time for write batch to database', default=30), diff --git a/monasca_persister/conf/kafka_metrics.py b/monasca_persister/conf/kafka_metrics.py index df900445356a..b30ead4a1485 100644 --- a/monasca_persister/conf/kafka_metrics.py +++ b/monasca_persister/conf/kafka_metrics.py @@ -38,6 +38,9 @@ kafka_metrics_opts = [ cfg.StrOpt('zookeeper_path', help='Path in zookeeper for kafka consumer group partitioning algorithm', default='/persister_partitions/$kafka_metrics.topic'), + cfg.IntOpt('batch_size', + help='Maximum number of metrics to buffer before writing to database', + default=20000), ] # Replace Default OPt with reference to kafka group option diff --git a/monasca_persister/repositories/cassandra/abstract_repository.py b/monasca_persister/repositories/cassandra/abstract_repository.py index a7257844ca4d..188f1e8386af 100644 --- a/monasca_persister/repositories/cassandra/abstract_repository.py +++ b/monasca_persister/repositories/cassandra/abstract_repository.py @@ -1,4 +1,5 @@ # (C) Copyright 2016 Hewlett Packard Enterprise Development Company LP +# (C) Copyright 2017 SUSE LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -12,26 +13,23 @@ # implied. # See the License for the specific language governing permissions and # limitations under the License. + import abc -from cassandra import cluster -from cassandra import query from oslo_config import cfg import six from monasca_persister.repositories import abstract_repository +from monasca_persister.repositories.cassandra import connection_util +conf = cfg.CONF @six.add_metaclass(abc.ABCMeta) class AbstractCassandraRepository(abstract_repository.AbstractRepository): - def __init__(self): super(AbstractCassandraRepository, self).__init__() - self.conf = cfg.CONF - - self._cassandra_cluster = cluster.Cluster( - self.conf.cassandra.cluster_ip_addresses.split(',')) - - self.cassandra_session = self._cassandra_cluster.connect( - self.conf.cassandra.keyspace) - self._batch_stmt = query.BatchStatement() + self._cluster = connection_util.create_cluster() + self._session = connection_util.create_session(self._cluster) + self._retention = conf.cassandra.retention_policy * 24 * 3600 + self._cache_size = conf.cassandra.max_definition_cache_size + self._max_batches = conf.cassandra.max_batches diff --git a/monasca_persister/repositories/cassandra/alarm_state_history_repository.py b/monasca_persister/repositories/cassandra/alarm_state_history_repository.py index 335f72bb1ec3..f9e72983081d 100644 --- a/monasca_persister/repositories/cassandra/alarm_state_history_repository.py +++ b/monasca_persister/repositories/cassandra/alarm_state_history_repository.py @@ -1,4 +1,5 @@ # (C) Copyright 2016 Hewlett Packard Enterprise Development Company LP +# (C) Copyright 2017 SUSE LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -12,9 +13,10 @@ # implied. # See the License for the specific language governing permissions and # limitations under the License. -import json +import ujson as json -from cassandra import query +from cassandra.concurrent import execute_concurrent_with_args +from oslo_config import cfg from oslo_log import log from monasca_persister.repositories.cassandra import abstract_repository @@ -22,51 +24,38 @@ from monasca_persister.repositories.utils import parse_alarm_state_hist_message LOG = log.getLogger(__name__) +UPSERT_CQL = ('update monasca.alarm_state_history USING TTL ? ' + 'set metric = ?, old_state = ?, new_state = ?, sub_alarms = ?, reason = ?, reason_data = ? ' + 'where tenant_id = ? and alarm_id = ? and time_stamp = ?') -class AlarmStateHistCassandraRepository( - abstract_repository.AbstractCassandraRepository): +class AlarmStateHistCassandraRepository(abstract_repository.AbstractCassandraRepository): def __init__(self): - super(AlarmStateHistCassandraRepository, self).__init__() - self._insert_alarm_state_hist_stmt = self.cassandra_session.prepare( - 'insert into alarm_state_history (tenant_id, alarm_id, ' - 'metrics, new_state, ' - 'old_state, reason, reason_data, ' - 'sub_alarms, time_stamp) values (?,?,?,?,?,?,?,?,?)') + self._upsert_stmt = self._session.prepare(UPSERT_CQL) def process_message(self, message): - - (alarm_id, metrics, new_state, old_state, link, - lifecycle_state, state_change_reason, - sub_alarms_json_snake_case, tenant_id, - time_stamp) = parse_alarm_state_hist_message( - message) - - alarm_state_hist = ( - tenant_id.encode('utf8'), - alarm_id.encode('utf8'), - json.dumps(metrics, ensure_ascii=False).encode( - 'utf8'), - new_state.encode('utf8'), - old_state.encode('utf8'), - state_change_reason.encode('utf8'), - "{}".encode('utf8'), - sub_alarms_json_snake_case.encode('utf8'), - time_stamp - ) - - LOG.debug(alarm_state_hist) + (alarm_id, metrics, new_state, old_state, link, lifecycle_state, state_change_reason, + sub_alarms_json_snake_case, + tenant_id, time_stamp) = parse_alarm_state_hist_message(message) + + alarm_state_hist = (self._retention, + json.dumps(metrics, ensure_ascii=False).encode('utf8'), + old_state.encode('utf8'), + new_state.encode('utf8'), + sub_alarms_json_snake_case.encode('utf8'), + state_change_reason.encode('utf8'), + "{}".encode('utf8'), + tenant_id.encode('utf8'), + alarm_id.encode('utf8'), + time_stamp) return alarm_state_hist def write_batch(self, alarm_state_hists): - - for alarm_state_hist in alarm_state_hists: - self._batch_stmt.add(self._insert_alarm_state_hist_stmt, - alarm_state_hist) - - self.cassandra_session.execute(self._batch_stmt) - - self._batch_stmt = query.BatchStatement() + while alarm_state_hists: + num_rows = min(len(alarm_state_hists), cfg.CONF.kafka_alarm_history.batch_size) + batch = alarm_state_hists[:num_rows] + execute_concurrent_with_args(self._session, self._upsert_stmt, batch) + alarm_state_hists = alarm_state_hists[num_rows:] diff --git a/monasca_persister/repositories/cassandra/connection_util.py b/monasca_persister/repositories/cassandra/connection_util.py new file mode 100644 index 000000000000..24a0c95e96c0 --- /dev/null +++ b/monasca_persister/repositories/cassandra/connection_util.py @@ -0,0 +1,51 @@ +# (C) Copyright 2017 SUSE LLC +# +# 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. + +from oslo_config import cfg +from cassandra.auth import PlainTextAuthProvider +from cassandra.cluster import Cluster +from cassandra.cluster import ConsistencyLevel +from cassandra.cluster import DCAwareRoundRobinPolicy +from cassandra.cluster import TokenAwarePolicy + +from monasca_persister.repositories.cassandra.retry_policy import MonascaRetryPolicy + +conf = cfg.CONF + +def create_cluster(): + user = conf.cassandra.user + if user: + auth_provider = PlainTextAuthProvider(username=user, password=conf.cassandra.password) + else: + auth_provider = None + + contact_points = [ip.dest for ip in conf.cassandra.contact_points] + cluster = Cluster(contact_points, + port=conf.cassandra.port, + auth_provider=auth_provider, + connect_timeout=conf.cassandra.connection_timeout, + load_balancing_policy=TokenAwarePolicy( + DCAwareRoundRobinPolicy(local_dc=conf.cassandra.local_data_center)), + ) + cluster.default_retry_policy = MonascaRetryPolicy(1, conf.cassandra.max_write_retries, + conf.cassandra.max_write_retries) + return cluster + + +def create_session(cluster): + session = cluster.connect(conf.cassandra.keyspace) + session.default_timeout = conf.cassandra.read_timeout + session.default_consistency_level = ConsistencyLevel.name_to_value[conf.cassandra.consistency_level] + return session diff --git a/monasca_persister/repositories/cassandra/metric_batch.py b/monasca_persister/repositories/cassandra/metric_batch.py new file mode 100644 index 000000000000..b6740db6902f --- /dev/null +++ b/monasca_persister/repositories/cassandra/metric_batch.py @@ -0,0 +1,152 @@ +# (C) Copyright 2017 SUSE LLC +# +# 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. + + +from oslo_log import log + +from cassandra.query import BatchStatement +from cassandra.query import BatchType + +LOG = log.getLogger(__name__) + + +class MetricBatch(object): + def __init__(self, metadata, load_balance_policy, batch_limit): + + self.metadata = metadata + self.batch_limit = batch_limit + self.lb_policy = load_balance_policy + self.metric_queries = dict() + self.dimension_queries = dict() + self.dimension_metric_queries = dict() + self.metric_dimension_queries = dict() + self.measurement_queries = dict() + + def batch_query_by_token(self, bound_stmt, query_map): + token = self.metadata.token_map.token_class.from_key(bound_stmt.routing_key) + + queue = query_map.get(token, None) + if not queue: + queue = [] + batch = BatchStatement(BatchType.UNLOGGED) + batch.add(bound_stmt) + queue.append((batch, Counter(1))) + query_map[token] = queue + else: + (batch, counter) = queue[-1] + if counter.value() < self.batch_limit: + batch.add(bound_stmt) + counter.increment() + else: + batch = BatchStatement(BatchType.UNLOGGED) + batch.add(bound_stmt) + queue.append((batch, Counter(1))) + + def add_metric_query(self, bound_stmt): + self.batch_query_by_token(bound_stmt, self.metric_queries) + + def add_dimension_query(self, bound_stmt): + self.batch_query_by_token(bound_stmt, self.dimension_queries) + + def add_dimension_metric_query(self, bound_stmt): + self.batch_query_by_token(bound_stmt, self.dimension_metric_queries) + + def add_metric_dimension_query(self, bound_stmt): + self.batch_query_by_token(bound_stmt, self.metric_dimension_queries) + + def add_measurement_query(self, bound_stmt): + self.batch_query_by_replicas(bound_stmt, self.measurement_queries) + + def batch_query_by_replicas(self, bound_stmt, query_map): + hosts = tuple(self.lb_policy.make_query_plan(working_keyspace=bound_stmt.keyspace, query=bound_stmt)) + + queue = query_map.get(hosts, None) + if not queue: + queue = [] + batch = BatchStatement(BatchType.UNLOGGED) + batch.add(bound_stmt) + queue.append((batch, Counter(1))) + query_map[hosts] = queue + else: + (batch, counter) = queue[-1] + if counter.value() < 30: + batch.add(bound_stmt) + counter.increment() + else: + batch = BatchStatement(BatchType.UNLOGGED) + batch.add(bound_stmt) + queue.append((batch, Counter(1))) + + def clear(self): + self.metric_queries.clear() + self.dimension_queries.clear() + self.dimension_metric_queries.clear() + self.metric_dimension_queries.clear() + self.measurement_queries.clear() + + @staticmethod + def log_token_batch_map(name, query_map): + LOG.info('%s : Size: %s; Tokens: |%s|' % (name, len(query_map), + '|'.join(['%s: %s' % ( + token, + ','.join([str(counter.value()) for (batch, counter) in queue])) + for token, queue in query_map.items()]))) + + @staticmethod + def log_replica_batch_map(name, query_map): + LOG.info('%s : Size: %s; Replicas: |%s|' % (name, len(query_map), '|'.join([ + '%s: %s' % ( + ','.join([h.address for h in hosts]), ','.join([str(counter.value()) for (batch, counter) in queue])) + for hosts, queue in query_map.items()]))) + + def get_all_batches(self): + self.log_token_batch_map("metric batches", self.metric_queries) + self.log_token_batch_map("dimension batches", self.dimension_queries) + self.log_token_batch_map("dimension metric batches", self.dimension_metric_queries) + self.log_token_batch_map("metric dimension batches", self.metric_dimension_queries) + self.log_replica_batch_map("measurement batches", self.measurement_queries) + + result_list = [] + + for q in self.measurement_queries.values(): + result_list.extend(q) + + for q in self.metric_queries.values(): + result_list.extend(q) + + for q in self.dimension_queries.values(): + result_list.extend(q) + + for q in self.dimension_metric_queries.values(): + result_list.extend(q) + + for q in self.metric_dimension_queries.values(): + result_list.extend(q) + + return result_list + + +class Counter(object): + def __init__(self, init_value=0): + self._count = init_value + + def increment(self): + self._count += 1 + + def increment_by(self, increment): + self._count += increment + + def value(self): + return self._count diff --git a/monasca_persister/repositories/cassandra/metrics_repository.py b/monasca_persister/repositories/cassandra/metrics_repository.py index ce10c094470c..2f5cf8fa392c 100644 --- a/monasca_persister/repositories/cassandra/metrics_repository.py +++ b/monasca_persister/repositories/cassandra/metrics_repository.py @@ -1,4 +1,5 @@ # (C) Copyright 2016 Hewlett Packard Enterprise Development Company LP +# (C) Copyright 2017 SUSE LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -12,105 +13,262 @@ # implied. # See the License for the specific language governing permissions and # limitations under the License. + +from __future__ import with_statement +from cachetools import LRUCache +from collections import namedtuple import hashlib -import json +import threading +import ujson as json + +from cassandra.concurrent import execute_concurrent -from cassandra import query from oslo_log import log -import urllib from monasca_persister.repositories.cassandra import abstract_repository +from monasca_persister.repositories.cassandra import token_range_query_manager +from monasca_persister.repositories.cassandra.metric_batch import MetricBatch from monasca_persister.repositories.utils import parse_measurement_message LOG = log.getLogger(__name__) +MEASUREMENT_INSERT_CQL = ('update monasca.measurements USING TTL ? ' + 'set value = ?, value_meta = ?, region = ?, tenant_id = ?, metric_name = ?, dimensions = ? ' + 'where metric_id = ? and time_stamp = ?') -class MetricCassandraRepository( - abstract_repository.AbstractCassandraRepository): +MEASUREMENT_UPDATE_CQL = ('update monasca.measurements USING TTL ? ' + 'set value = ?, value_meta = ? where metric_id = ? and time_stamp = ?') - def __init__(self): +METRICS_INSERT_CQL = ('update monasca.metrics USING TTL ? ' + 'set metric_id = ?, created_at = ?, updated_at = ? ' + 'where region = ? and tenant_id = ? and metric_name = ? and dimensions = ? ' + 'and dimension_names = ?') - super(MetricCassandraRepository, self).__init__() +METRICS_UPDATE_CQL = ('update monasca.metrics USING TTL ? ' + 'set updated_at = ? ' + 'where region = ? and tenant_id = ? and metric_name = ? and dimensions = ? ' + 'and dimension_names = ?') - self._insert_measurement_stmt = self.cassandra_session.prepare( - 'insert into measurements (tenant_id,' - 'region, metric_hash, time_stamp, value,' - 'value_meta) values (?, ?, ?, ?, ?, ?)') +DIMENSION_INSERT_CQL = ('insert into monasca.dimensions ' + '(region, tenant_id, name, value) values (?, ?, ?, ?)') - self._insert_metric_map_stmt = self.cassandra_session.prepare( - 'insert into metric_map (tenant_id,' - 'region, metric_hash, ' - 'metric_map) values' - '(?,?,?,?)') +DIMENSION_METRIC_INSERT_CQL = ('insert into monasca.dimensions_metrics ' + '(region, tenant_id, dimension_name, dimension_value, metric_name) ' + 'values (?, ?, ?, ?, ?)') - def process_message(self, message): +METRIC_DIMENSION_INSERT_CQL = ('insert into monasca.metrics_dimensions ' + '(region, tenant_id, metric_name, dimension_name, dimension_value) ' + 'values (?, ?, ?, ?, ?)') - (dimensions, metric_name, region, tenant_id, time_stamp, value, - value_meta) = parse_measurement_message(message) +RETRIEVE_DIMENSION_CQL = 'select region, tenant_id, name, value from dimensions' - metric_hash, metric_map = create_metric_hash(metric_name, - dimensions) +RETRIEVE_METRIC_DIMENSION_CQL = ('select region, tenant_id, metric_name, ' + 'dimension_name, dimension_value from metrics_dimensions ' + 'WHERE token(region, tenant_id, metric_name) > ? ' + 'and token(region, tenant_id, metric_name) <= ? ') - measurement = (tenant_id.encode('utf8'), - region.encode('utf8'), - metric_hash, - time_stamp, - value, - json.dumps(value_meta, ensure_ascii=False).encode( - 'utf8')) +Metric = namedtuple('Metric', ['id', 'region', 'tenant_id', 'name', 'dimension_list', 'dimension_names', + 'time_stamp', 'value', 'value_meta']) - LOG.debug(measurement) - return MetricMeasurementInfo( - tenant_id.encode('utf8'), - region.encode('utf8'), - metric_hash, - metric_map, - measurement) +class MetricCassandraRepository(abstract_repository.AbstractCassandraRepository): + def __init__(self): + super(MetricCassandraRepository, self).__init__() + + self._lock = threading.RLock() + + LOG.debug("prepare cql statements...") + + self._measurement_insert_stmt = self._session.prepare(MEASUREMENT_INSERT_CQL) + self._measurement_insert_stmt.is_idempotent = True - def write_batch(self, metric_measurement_infos): + self._measurement_update_stmt = self._session.prepare(MEASUREMENT_UPDATE_CQL) + self._measurement_update_stmt.is_idempotent = True - for metric_measurement_info in metric_measurement_infos: + self._metric_insert_stmt = self._session.prepare(METRICS_INSERT_CQL) + self._metric_insert_stmt.is_idempotent = True - self._batch_stmt.add(self._insert_measurement_stmt, - metric_measurement_info.measurement) + self._metric_update_stmt = self._session.prepare(METRICS_UPDATE_CQL) + self._metric_update_stmt.is_idempotent = True - metric_map = (metric_measurement_info.tenant_id, - metric_measurement_info.region, - metric_measurement_info.metric_hash, - metric_measurement_info.metric_map) + self._dimension_stmt = self._session.prepare(DIMENSION_INSERT_CQL) + self._dimension_stmt.is_idempotent = True - self._batch_stmt.add(self._insert_metric_map_stmt, - metric_map) + self._dimension_metric_stmt = self._session.prepare(DIMENSION_METRIC_INSERT_CQL) + self._dimension_metric_stmt.is_idempotent = True - self.cassandra_session.execute(self._batch_stmt) + self._metric_dimension_stmt = self._session.prepare(METRIC_DIMENSION_INSERT_CQL) + self._metric_dimension_stmt.is_idempotent = True - self._batch_stmt = query.BatchStatement() + self._retrieve_metric_dimension_stmt = self._session.prepare(RETRIEVE_METRIC_DIMENSION_CQL) + self._metric_batch = MetricBatch(self._cluster.metadata, self._cluster.load_balancing_policy, self._max_batches) -class MetricMeasurementInfo(object): + self._metric_id_cache = LRUCache(self._cache_size) + self._dimension_cache = LRUCache(self._cache_size) + self._metric_dimension_cache = LRUCache(self._cache_size) - def __init__(self, tenant_id, region, metric_hash, metric_map, - measurement): + self._load_dimension_cache() + self._load_metric_dimension_cache() - self.tenant_id = tenant_id - self.region = region - self.metric_hash = metric_hash - self.metric_map = metric_map - self.measurement = measurement + def process_message(self, message): + (dimensions, metric_name, region, tenant_id, time_stamp, value, + value_meta) = parse_measurement_message(message) + with self._lock: + dim_names = [] + dim_list = [] + for name in sorted(dimensions.iterkeys()): + dim_list.append('%s\t%s' % (name, dimensions[name])) + dim_names.append(name) + + hash_string = '%s\0%s\0%s\0%s' % (region, tenant_id, metric_name, '\0'.join(dim_list)) + metric_id = hashlib.sha1(hash_string.encode('utf8')).hexdigest() + + metric = Metric(id=metric_id, + region=region, + tenant_id=tenant_id, + name=metric_name, + dimension_list=dim_list, + dimension_names=dim_names, + time_stamp=time_stamp, + value=value, + value_meta=json.dumps(value_meta, ensure_ascii=False)) + + id_bytes = bytearray.fromhex(metric.id) + if self._metric_id_cache.get(metric.id, None): + measurement_bound_stmt = self._measurement_update_stmt.bind((self._retention, + metric.value, + metric.value_meta, + id_bytes, + metric.time_stamp)) + self._metric_batch.add_measurement_query(measurement_bound_stmt) + + metric_update_bound_stmt = self._metric_update_stmt.bind((self._retention, + metric.time_stamp, + metric.region, + metric.tenant_id, + metric.name, + metric.dimension_list, + metric.dimension_names)) + self._metric_batch.add_metric_query(metric_update_bound_stmt) + + return metric + + self._metric_id_cache[metric.id] = metric.id + + metric_insert_bound_stmt = self._metric_insert_stmt.bind((self._retention, + id_bytes, + metric.time_stamp, + metric.time_stamp, + metric.region, + metric.tenant_id, + metric.name, + metric.dimension_list, + metric.dimension_names)) + self._metric_batch.add_metric_query(metric_insert_bound_stmt) + + for dim in metric.dimension_list: + (name, value) = dim.split('\t') + dim_key = self._get_dimnesion_key(metric.region, metric.tenant_id, name, value) + if not self._dimension_cache.get(dim_key, None): + dimension_bound_stmt = self._dimension_stmt.bind((metric.region, + metric.tenant_id, + name, + value)) + self._metric_batch.add_dimension_query(dimension_bound_stmt) + self._dimension_cache[dim_key] = dim_key + + metric_dim_key = self._get_metric_dimnesion_key(metric.region, metric.tenant_id, metric.name, name, + value) + if not self._metric_dimension_cache.get(metric_dim_key, None): + dimension_metric_bound_stmt = self._dimension_metric_stmt.bind((metric.region, + metric.tenant_id, + name, + value, + metric.name)) + self._metric_batch.add_dimension_metric_query(dimension_metric_bound_stmt) + + metric_dimension_bound_stmt = self._metric_dimension_stmt.bind((metric.region, + metric.tenant_id, + metric.name, + name, + value)) + self._metric_batch.add_metric_dimension_query(metric_dimension_bound_stmt) + + self._metric_dimension_cache[metric_dim_key] = metric_dim_key + + measurement_insert_bound_stmt = self._measurement_insert_stmt.bind((self._retention, + metric.value, + metric.value_meta, + metric.region, + metric.tenant_id, + metric.name, + metric.dimension_list, + id_bytes, + metric.time_stamp)) + self._metric_batch.add_measurement_query(measurement_insert_bound_stmt) + + return metric + + def write_batch(self, metrics): + + with self._lock: + batch_list = self._metric_batch.get_all_batches() + + results = execute_concurrent(self._session, batch_list, raise_on_first_error=True) + + self._handle_results(results) + + self._metric_batch.clear() + + LOG.info("flushed %s metrics", len(metrics)) + + @staticmethod + def _handle_results(results): + for (success, result) in results: + if not success: + raise result + + def _load_dimension_cache(self): + + rows = self._session.execute(RETRIEVE_DIMENSION_CQL) + + if not rows: + return + + for row in rows: + key = self._get_dimnesion_key(row.region, row.tenant_id, row.name, row.value) + self._dimension_cache[key] = key + + LOG.info("loaded %s dimension entries cache from database into cache." % self._dimension_cache.currsize) + + @staticmethod + def _get_dimnesion_key(region, tenant_id, name, value): + return '%s\0%s\0%s\0%s' % (region, tenant_id, name, value) + + def _load_metric_dimension_cache(self): + qm = token_range_query_manager.TokenRangeQueryManager(RETRIEVE_METRIC_DIMENSION_CQL, + self._process_metric_dimension_query) -def create_metric_hash(metric_name, dimensions): + token_ring = self._cluster.metadata.token_map.ring - dimensions['__name__'] = urllib.quote_plus(metric_name) + qm.query(token_ring) - hash_string = '' + def _process_metric_dimension_query(self, rows): + + cnt = 0 + for row in rows: + key = self._get_metric_dimnesion_key(row.region, row.tenant_id, row.metric_name, row.dimension_name, + row.dimension_value) + self._metric_dimension_cache[key] = key + cnt += 1 - for dim_name in sorted(dimensions.iterkeys()): - dimension = (urllib.quote_plus(dim_name) + '=' + urllib.quote_plus( - dimensions[dim_name])) - hash_string += dimension + LOG.info("loaded %s metric dimension entries from database into cache." % cnt) + LOG.info( + "total loaded %s metric dimension entries in cache." % self._metric_dimension_cache.currsize) - sha1_hash = hashlib.sha1(hash_string).hexdigest() + @staticmethod + def _get_metric_dimnesion_key(region, tenant_id, metric_name, dimension_name, dimension_value): - return bytearray.fromhex(sha1_hash), dimensions + return '%s\0%s\0%s\0%s\0%s' % (region, tenant_id, metric_name, dimension_name, dimension_value) diff --git a/monasca_persister/repositories/cassandra/retry_policy.py b/monasca_persister/repositories/cassandra/retry_policy.py new file mode 100644 index 000000000000..163337d6b9e9 --- /dev/null +++ b/monasca_persister/repositories/cassandra/retry_policy.py @@ -0,0 +1,49 @@ +# (C) Copyright 2017 SUSE LLC +# +# 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. + +from cassandra.policies import RetryPolicy + + +class MonascaRetryPolicy(RetryPolicy): + + def __init__(self, read_attempts, write_attempts, unavailable_attempts): + + super(MonascaRetryPolicy, self).__init__() + + self.read_attempts = read_attempts + self.write_attempts = write_attempts + self.unavailable_attempts = unavailable_attempts + + def on_read_timeout(self, query, consistency, required_responses, + received_responses, data_retrieved, retry_num): + + if retry_num >= self.read_attempts: + return self.RETHROW, None + elif received_responses >= required_responses and not data_retrieved: + return self.RETRY, consistency + else: + return self.RETHROW, None + + def on_write_timeout(self, query, consistency, write_type, + required_responses, received_responses, retry_num): + + if retry_num >= self.write_attempts: + return self.RETHROW, None + else: + return self.RETRY, consistency + + def on_unavailable(self, query, consistency, required_replicas, alive_replicas, retry_num): + + return (self.RETRY_NEXT_HOST, consistency) if retry_num < self.unavailable_attempts else (self.RETHROW, None) diff --git a/monasca_persister/repositories/cassandra/token_range_query_manager.py b/monasca_persister/repositories/cassandra/token_range_query_manager.py new file mode 100644 index 000000000000..e90a47fab91a --- /dev/null +++ b/monasca_persister/repositories/cassandra/token_range_query_manager.py @@ -0,0 +1,67 @@ +# (C) Copyright 2017 SUSE LLC +# +# 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. + +import multiprocessing + +from oslo_config import cfg +from oslo_log import log + +from monasca_persister.repositories.cassandra import connection_util + + +LOG = log.getLogger(__name__) + +conf = cfg.CONF + + +class TokenRangeQueryManager(object): + def __init__(self, cql, result_handler, process_count=None): + if process_count: + self._process_count = process_count + else: + self._process_count = multiprocessing.cpu_count() + + self._pool = multiprocessing.Pool(processes=self._process_count, initializer=self._setup, + initargs=(cql, result_handler,)) + + @classmethod + def _setup(cls, cql, result_handler): + cls.cluster = connection_util.create_cluster() + cls.session = connection_util.create_session(cls.cluster) + cls.prepared = cls.session.prepare(cql) + cls.result_handler = result_handler + + def close_pool(self): + self._pool.close() + self._pool.join() + + def query(self, token_ring): + + range_size = len(token_ring) / self._process_count + 1 + start_index = 0 + params = [] + while start_index < len(token_ring): + end_index = start_index + range_size - 1 + if end_index >= len(token_ring): + end_index = len(token_ring) - 1 + params.append((token_ring[start_index].value, token_ring[end_index].value)) + start_index = end_index + 1 + + self._pool.map(execute_query_token_range, params, 1) + + +def execute_query_token_range(token_range): + results = TokenRangeQueryManager.session.execute(TokenRangeQueryManager.prepared.bind(token_range)) + TokenRangeQueryManager.result_handler(results) diff --git a/monasca_persister/repositories/persister.py b/monasca_persister/repositories/persister.py index d1a497514bea..51c21523d258 100644 --- a/monasca_persister/repositories/persister.py +++ b/monasca_persister/repositories/persister.py @@ -1,4 +1,5 @@ # (C) Copyright 2016 Hewlett Packard Enterprise Development Company LP +# (C) Copyright 2017 SUSE LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -29,7 +30,7 @@ class Persister(object): self._kafka_topic = kafka_conf.topic - self._database_batch_size = kafka_conf.database_batch_size + self._batch_size = kafka_conf.batch_size self._consumer = consumer.KafkaConsumer( kafka_conf.uri, @@ -71,7 +72,7 @@ class Persister(object): LOG.exception('Error processing message. Message is ' 'being dropped. {}'.format(message)) - if len(self._data_points) >= self._database_batch_size: + if len(self._data_points) >= self._batch_size: self._flush() except Exception: LOG.exception(
Locations
Projects
Search
Status Monitor
Help
OpenBuildService.org
Documentation
API Documentation
Code of Conduct
Contact
Support
@OBShq
Terms
openSUSE Build Service is sponsored by
The Open Build Service is an
openSUSE project
.
Sign Up
Log In
Places
Places
All Projects
Status Monitor
