StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: Redis caching

Browse Source
This commit is contained in:
Igor Artamonov
2020-03-14 22:25:27 -04:00
parent 67ba1457fa
commit 35cb35fe0a
23 changed files with 529 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/kotlin/io/emeraldpay/dshackle/Config.kt
View File

@@ -19,25 +19,22 @@ import com.fasterxml.jackson.core.Version
import com.fasterxml.jackson.databind.DeserializationFeature
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.module.SimpleModule
import io.lettuce.core.AbstractRedisClient
import io.lettuce.core.cluster.RedisClusterClient
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.context.ApplicationContext
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.context.annotation.Import
import org.springframework.core.env.Environment
import org.springframework.scheduling.annotation.EnableAsync
import org.springframework.scheduling.annotation.EnableScheduling
import org.springframework.scheduling.annotation.Scheduled
import reactor.core.scheduler.Scheduler
import reactor.core.scheduler.Schedulers
import java.io.File
import java.lang.IllegalStateException
import java.text.SimpleDateFormat
import java.util.*
import java.util.concurrent.Executors
import kotlin.system.exitProcess
@Configuration
@EnableScheduling
@@ -82,4 +79,9 @@ open class Config(
open fun fileResolver(): FileResolver {
return FileResolver(configDir())
}
@Bean
open fun redisClient(): AbstractRedisClient {
return RedisClusterClient.create("redis://password@localhost:6379/0");
}
}

26
src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt vendored
View File

@@ -24,8 +24,9 @@ class BlocksRedisCache(
companion object {
private val log = LoggerFactory.getLogger(BlocksRedisCache::class.java)
// max caching time is 24 hours
private const val MAX_CACHE_TIME_HOURS = 24L
private const val MAX_CACHE_TIME_MINUTES = 60L
// doesn't make sense to cached in redis short living objects
private const val MIN_CACHE_TIME_SECONDS = 10
}
override fun read(key: BlockHash): Mono<BlockJson<TransactionRefJson>> {
@@ -37,23 +38,36 @@ class BlocksRedisCache(
}
}
fun evict(id: BlockHash): Mono<Void> {
return Mono.just(id)
.flatMap {
redis.del(key(it))
}
.then()
}
/**
* Add to cache.
* Note that it returns Mono<Void> which must be subscribed to actually save
*/
open fun add(block: BlockJson<TransactionRefJson>): Mono<Void> {
fun add(block: BlockJson<TransactionRefJson>): Mono<Void> {
if (block.timestamp == null || block.hash == null) {
return Mono.empty()
}
return Mono.just(block)
.flatMap { block ->
val data = objectMapper.writeValueAsString(block)
//default caching time is age of the block, i.e. block create hour ago
//keep for hour, but block create 10 seconds ago cache for 10 seconds, as it
//still can be replaced in the blockchain
val age = Instant.now().epochSecond - block.timestamp.epochSecond
val ttl = min(age, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS))
redis.setex(key(block.hash), ttl, data)
val ttl = min(age, TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES))
if (ttl > MIN_CACHE_TIME_SECONDS) {
redis.setex(key(block.hash), ttl, data)
} else {
Mono.empty()
}
}
.doOnError {
log.warn("Failed to save to Redis: ${it.message}")
@@ -68,7 +82,7 @@ class BlocksRedisCache(
/**
* Key in Redis
*/
open fun key(hash: BlockHash): String {
fun key(hash: BlockHash): String {
return "block:${chain.id}:${hash.toHex()}"
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksWithTxCache.kt vendored
View File

@@ -2,8 +2,10 @@ package io.emeraldpay.dshackle.cache
import io.emeraldpay.dshackle.reader.Reader
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import org.springframework.beans.BeanUtils
import reactor.core.publisher.Flux
@@ -17,8 +19,8 @@ import reactor.core.publisher.Mono
* If any of the expected block transactions is not available it returns empty
*/
class BlocksWithTxCache(
private val blocks: BlocksMemCache,
private val txes: TxMemCache
private val blocks: Reader<BlockHash, BlockJson<TransactionRefJson>>,
private val txes: Reader<TransactionId, TransactionJson>
): Reader<BlockHash, BlockJson<TransactionJson>> {
companion object {

77
src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt vendored
View File

@@ -1,5 +1,6 @@
package io.emeraldpay.dshackle.cache
import io.emeraldpay.dshackle.reader.CompoundReader
import io.emeraldpay.dshackle.reader.Reader
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
@@ -7,11 +8,16 @@ import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.TopicProcessor
open class Caches(
private val blocksByHash: BlocksMemCache,
private val memBlocksByHash: BlocksMemCache,
private val blocksByHeight: HeightCache,
private val txsByHash: TxMemCache
private val memTxsByHash: TxMemCache,
private val redisBlocksByHash: BlocksRedisCache?,
private val redisTxsByHash: TxRedisCache?
) {
companion object {
@@ -28,6 +34,22 @@ open class Caches(
}
}
private val blocksByHash: Reader<BlockHash, BlockJson<TransactionRefJson>>
private val txsByHash: Reader<TransactionId, TransactionJson>
init {
blocksByHash = if (redisBlocksByHash == null) {
memBlocksByHash
} else {
CompoundReader(memBlocksByHash, redisBlocksByHash)
}
txsByHash = if (redisTxsByHash == null) {
memTxsByHash
} else {
CompoundReader(memTxsByHash, redisTxsByHash)
}
}
/**
* Cache data that was just requested
*/
@@ -40,32 +62,51 @@ open class Caches(
}
fun cache(tag: Tag, tx: TransactionJson) {
txsByHash.add(tx)
//do not cache transactions that are not in a block yet
if (tx.blockHash == null) {
return
}
memTxsByHash.add(tx)
memBlocksByHash.get(tx.blockHash)?.let { block ->
redisTxsByHash?.add(tx, block)
}
}
fun cache(tag: Tag, block: BlockJson<TransactionRefJson>) {
val job = ArrayList<Mono<Void>>()
if (tag == Tag.LATEST) {
blocksByHash.add(block)
//for LATEST data cache in memory, it will be short living so better to avoid Redis
memBlocksByHash.add(block)
val replaced = blocksByHeight.add(block)
//evict cached transactions if an existing block was updated
replaced?.let { replacedBlockHash ->
var evicted = false
blocksByHash.get(replacedBlockHash)?.let { block ->
txsByHash.evict(block)
redisBlocksByHash?.evict(replacedBlockHash)
memBlocksByHash.get(replacedBlockHash)?.let { block ->
memTxsByHash.evict(block)
redisTxsByHash?.evict(block)
evicted = true
}
if (!evicted) {
txsByHash.evict(replacedBlockHash)
memTxsByHash.evict(replacedBlockHash)
}
}
} else if (tag == Tag.REQUESTED) {
// if block with transactions was requests cache only transactions
block.transactions.forEach { tx ->
if (tx is TransactionJson) {
cache(Tag.REQUESTED, tx)
//shouldn't cache block json with transactions, separate txes and blocks with refs
val blockOnly = block.withoutTransactionDetails()
memBlocksByHash.add(blockOnly)
redisBlocksByHash?.add(blockOnly)?.let(job::add)
// now cache only transactions
val transactions = block.transactions.filterIsInstance<TransactionJson>()
if (transactions.isNotEmpty()) {
transactions.forEach { cache(Tag.REQUESTED, it) }
if (redisTxsByHash != null) {
job.add(Flux.fromIterable(transactions).flatMap { redisTxsByHash.add(it, block) }.then())
}
}
}
Flux.fromIterable(job).flatMap { it }.subscribe() //TODO move out to a caller
}
fun getBlocksByHash(): Reader<BlockHash, BlockJson<TransactionRefJson>> {
@@ -107,12 +148,19 @@ open class Caches(
private var blocksByHash: BlocksMemCache? = null
private var blocksByHeight: HeightCache? = null
private var txsByHash: TxMemCache? = null
private var redisBlocksByHash: BlocksRedisCache? = null
private var redisTxsByHash: TxRedisCache? = null
fun setBlockByHash(cache: BlocksMemCache): Builder {
blocksByHash = cache
return this
}
fun setBlockByHash(cache: BlocksRedisCache): Builder {
redisBlocksByHash = cache
return this
}
fun setBlockByHeight(cache: HeightCache): Builder {
blocksByHeight = cache
return this
@@ -123,6 +171,11 @@ open class Caches(
return this
}
fun setTxByHash(cache: TxRedisCache): Builder {
redisTxsByHash = cache
return this
}
fun build(): Caches {
if (blocksByHash == null) {
blocksByHash = BlocksMemCache()
@@ -133,7 +186,7 @@ open class Caches(
if (txsByHash == null) {
txsByHash = TxMemCache()
}
return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!)
return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash)
}
}
}

86
src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt vendored Normal file
View File

@@ -0,0 +1,86 @@
package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.config.EnvVariables
import io.emeraldpay.grpc.Chain
import io.lettuce.core.RedisClient
import io.lettuce.core.RedisURI
import io.lettuce.core.api.StatefulRedisConnection
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Value
import org.springframework.core.env.Environment
import org.springframework.stereotype.Repository
import java.util.*
import java.util.concurrent.ConcurrentHashMap
import javax.annotation.PostConstruct
import kotlin.collections.HashMap
@Repository
class CachesFactory(
@Autowired private val objectMapper: ObjectMapper,
@Autowired private val env: Environment
) {
companion object {
private val log = LoggerFactory.getLogger(CachesFactory::class.java)
private const val CONFIG_PREFIX = "cache.redis"
}
private var redis: StatefulRedisConnection<String, String>? = null
private val all = EnumMap<Chain, Caches>(io.emeraldpay.grpc.Chain::class.java)
@PostConstruct
fun init() {
if (!env.getProperty("${CONFIG_PREFIX}.enabled", Boolean::class.java, false)) {
return
}
val address = env.getProperty("${CONFIG_PREFIX}.host", "127.0.0.1")
val port = env.getProperty("${CONFIG_PREFIX}.port", Int::class.java, 6379)
var uri = RedisURI.builder()
.withHost(address)
.withPort(port)
env.getProperty("${CONFIG_PREFIX}.db", Int::class.java)?.let { value ->
uri = uri.withDatabase(value)
}
//log URI _before_ adding a password, to avoid leaking it to the log
log.info("Use Redis cache at: ${uri.build().toURI()}")
env.getProperty("${CONFIG_PREFIX}.password")?.let { value ->
uri = uri.withPassword(value)
}
val client = RedisClient.create(uri.build())
val ping = client.connect().sync().ping()
if (ping != "PONG") {
throw IllegalStateException("Redis connection is not configured. Response: $ping")
}
redis = client.connect()
}
private fun initCache(chain: Chain): Caches {
val caches = Caches.newBuilder()
redis?.let { redis ->
caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain, objectMapper))
caches.setTxByHash(TxRedisCache(redis.reactive(), chain, objectMapper))
}
return caches.build()
}
fun getCaches(chain: Chain): Caches {
val existing = all[chain]
if (existing == null) {
synchronized(all) {
if (!all.containsKey(chain)) {
all[chain] = initCache(chain)
}
}
return getCaches(chain)
}
return existing
}
}

13
src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt vendored
View File

@@ -39,7 +39,7 @@ class TxRedisCache(
}
}
open fun evict(block: BlockJson<TransactionRefJson>): Mono<Void> {
fun evict(block: BlockJson<TransactionRefJson>): Mono<Void> {
return Mono.just(block)
.map { block ->
block.transactions.map {
@@ -50,8 +50,15 @@ class TxRedisCache(
}.then()
}
fun evict(id: TransactionId): Mono<Void> {
return Mono.just(id)
.flatMap {
redis.del(key(it))
}
.then()
}
open fun add(tx: TransactionJson, block: BlockJson<TransactionRefJson>): Mono<Void> {
fun add(tx: TransactionJson, block: BlockJson<TransactionRefJson>): Mono<Void> {
if (tx.blockHash == null || block.hash == null || tx.blockHash != block.hash || block.timestamp == null) {
return Mono.empty()
}
@@ -78,7 +85,7 @@ class TxRedisCache(
/**
* Key in Redis
*/
open fun key(hash: TransactionId): String {
fun key(hash: TransactionId): String {
return "tx:${chain.id}:${hash.toHex()}"
}
}

19
src/main/kotlin/io/emeraldpay/dshackle/config/EnvVariables.kt Normal file
View File

@@ -0,0 +1,19 @@
package io.emeraldpay.dshackle.config
/**
* Update configuration value from environment variables. Format: ${ENV_VAR_NAME}
*/
class EnvVariables {
companion object {
private val envRegex = Regex("\\$\\{(\\w+?)}")
}
fun postProcess(value: String): String {
return envRegex.replace(value) { m ->
m.groups[1]?.let { g ->
System.getProperty(g.value) ?: System.getenv(g.value) ?: ""
} ?: ""
}
}
}

13
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -32,7 +32,7 @@ import java.time.Duration
class UpstreamsConfigReader {
private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java)
private val envRegex = Regex("\\$\\{(\\w+?)}")
private val envVariables = EnvVariables()
fun read(input: InputStream): UpstreamsConfig {
val yaml = Yaml()
@@ -268,13 +268,13 @@ class UpstreamsConfigReader {
private fun getListOfString(mappingNode: MappingNode?, key: String): List<String>? {
return getList<ScalarNode>(mappingNode, key)?.value
?.map { it.value }
?.map(this::postProcess)
?.map(envVariables::postProcess)
}
private fun getValueAsString(mappingNode: MappingNode?, key: String): String? {
return getValue(mappingNode, key)?.let {
return@let it.value
}?.let(this::postProcess)
}?.let(envVariables::postProcess)
}
private fun getValueAsInt(mappingNode: MappingNode?, key: String): Int? {
@@ -305,11 +305,4 @@ class UpstreamsConfigReader {
}
}
fun postProcess(value: String): String {
return envRegex.replace(value) { m ->
m.groups[1]?.let { g ->
System.getProperty(g.value) ?: System.getenv(g.value) ?: ""
} ?: ""
}
}
}

16
src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt
View File

@@ -15,24 +15,22 @@
*/
package io.emeraldpay.dshackle.reader
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
/**
* Composition of multiple readers. Reader returns first value returned by any of the source readers.
*/
class CompoundReader<K, D>(
private val readers: Collection<Reader<K, D>>
private vararg val readers: Reader<K, D>
): Reader<K, D> {
override fun read(key: K): Mono<D> {
if (readers.isEmpty()) {
return Mono.empty()
}
var result = readers.first().read(key)
if (readers.size == 1) {
return result
}
readers.stream().skip(1).forEach {
result = result.switchIfEmpty(it.read(key))
}
return result
return Flux.fromIterable(readers.asIterable())
.flatMap { it.read(key) }.next()
}
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt
View File

@@ -38,6 +38,9 @@ open class CachingEthereumApi(
companion object {
private val log = LoggerFactory.getLogger(CachingEthereumApi::class.java)
/**
* Create caching API with empty memory-only cache
*/
@JvmStatic
fun empty(): CachingEthereumApi {
return CachingEthereumApi(ObjectMapper(), Caches.default(), EmptyEthereumHead())

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt
View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.startup.UpstreamChange
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.QuorumBasedMethods
@@ -35,7 +36,8 @@ import kotlin.concurrent.withLock
@Repository
class CurrentUpstreams(
@Autowired private val objectMapper: ObjectMapper
@Autowired private val objectMapper: ObjectMapper,
@Autowired private val cachesFactory: CachesFactory
): Upstreams {
private val log = LoggerFactory.getLogger(CurrentUpstreams::class.java)
@@ -55,7 +57,7 @@ class CurrentUpstreams(
log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed")
} else {
if (current == null) {
val created = ChainUpstreams(chain, ArrayList<Upstream>(), Caches.default(), objectMapper)
val created = ChainUpstreams(chain, ArrayList<Upstream>(), cachesFactory.getCaches(chain), objectMapper)
if (up is CachesEnabled) {
up.setCaches(created.caches)
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt
View File

@@ -101,7 +101,12 @@ open class DirectEthereumApi(
return rpcClient.execute(callMapping(method, params))
.timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout")))
.doOnNext { value ->
caches?.cacheRequested(value)
try {
caches?.cacheRequested(value)
} catch (e: Throwable) {
//ignore all caching errors, client shouldn't have problems because of them
log.warn("Uncaught caching exception", e)
}
}
}