twitter / storehaus Goto Github PK

Really awesome:

http://about.travis-ci.org/docs/user/database-setup/

We can use this to actually test many of our services.

With a Bijection[K,(K1,K2)] then you can make a Bijection[(K1) => ReadableStore[K2,V], ReadableStore[K,V]] and similarly with a Store.

We often call this sharding (example: K is an Integer, K1 = K % N, K2 = K / N).

http://aws.amazon.com/dynamodb/

Clojure's core.cache has this to say:

"Cached values will be referred to with SoftReferences, allowing the values to be garbage collected when there is memory pressure on the JVM."

An implementation:

https://github.com/clojure/core.cache/blob/master/src/main/clojure/clojure/core/cache.clj#L327

Sort of a bug.

The notion of a VersionedStore would make it easy to create a distributed read-only database backed by storehaus stores that loaded themselves off of disk.

import com.twitter.util.{ Duration, Future, Timer }

/**
  * ReadableStore with some notion of versioning.
  */
trait VersionedStore[-K, +V] extends ReadableStore[K, (Long, V)] {
  def currentVersion: Option[Long]
}

/**
  * Fired when an updating versioned store is not yet loaded.
  */
class StoreNotLoadedException extends RuntimeException

/**
  * The typical not-yet-loaded versioned store. This store returns
  * exceptions for every get and multiGet.
  */
object ExceptionalStore extends VersionedStore[Any, Nothing] {
  override def get(k: Any) = Future.exception(new StoreNotLoadedException)
  override val currentVersion = None
}

class ConstantVersionedStore[-K, +V](store: ReadableStore[K, V], version: Long) extends VersionedStore[K, V] {
  override val currentVersion = Some(version)

  protected def prependVersion(f: Future[Option[V]], version: Long) =
    f.map { _.map { (version, _) } }

  override def get(k: K) = prependVersion(store.get(k), version)
  override def multiGet[K1 <: K](ks: Set[K1]): Map[K1, Future[Option[(Long, V)]]] =
    store.multiGet(ks).mapValues { prependVersion(_, version) }
}

/**
  * Versioned store implementation capable of updating itself to some
  * new version, potentially asynchronously. One use case here would a
  * store that loads its backing dataset across the network from
  * HDFS. Every time a new version appeared on HDFS the watcher
  * process would switch the backing store behind the scenes.
  */
class UpdatingVersionedStore[-K, +V](updateInterval: Duration)(watcherProcess: Option[Long] => Future[Option[VersionedStore[K, V]]])
  (implicit timer: Timer)
    extends VersionedStore[K, V] {
  protected var innerStore: Future[VersionedStore[K, V]] = Future.value(ExceptionalStore)

  timer.schedule(updateInterval) {
    innerStore = innerStore
      .join(watcherProcess(currentVersion))
      .map { case (oldStore, optNewStore) => optNewStore.getOrElse(oldStore) }
  }
  override def get(k: K) = innerStore.flatMap { _.get(k) }
  override def multiGet(ks: Set[K]) =
    FutureOps.liftFutureValues(ks, innerStore.map { _.multiGet(ks) })
}

Suppose your client is going to make a bunch of gets, but you want to make sure those are batched into blocks of at least N or after a period of at most T, and send it to multiGet.

This makes it easy for callers to continue to use the get API, but to be more efficient and block them up internally with a multiGet.

We can do the same thing on write/merge (in fact, we have some of this for merging now).

The convenience if using an Injection to convert ConvertedStores values has landed it in a somewhat awkward home, the algebra module. I get that we want to keep storehaus-core lean but there's some residual strangeness in that I can enrich my readable store with convert but my enriched stores have no such luck because the pairing of converting for stores is under the algebra source tree. As a user, I'd expect that when I can call convert on my extension of Store, I'd get back a Store. Instead I get back a ReadableStore.

I can take a stab at making this less awkward of anyone can point me in the right direction.

The store returned by the "first" implementation on the ReadableStore and Store objects currently returns the first non-exceptional value. Some other nice combinators would be:

• Return the fastest value, exceptional or non-exceptional (probably useless)
• Return the first non-None value.

I see DRPCStore living here, along with the DecoderBolt and the relevant piece of our SinkBolt.

DRPCStore, for reference (needs BatchID removed, of course)

https://gist.github.com/sritchie/4951204

package com.twitter.summingbird.storm.store

import backtype.storm.utils.DRPCClient
import com.twitter.bijection.Bijection
import com.twitter.util.{ Future, FuturePool }
import com.twitter.storehaus.ReadableStore
import com.twitter.summingbird.util.RpcBijection
import java.util.concurrent.Executors

import Bijection.asMethod // enable "as" syntax

/**
 * Wrapper over backtype.storm.utils.DRPCClient.
 * This ReadableStore allows the user to perform online read-only
 * queries through Storm's DRPC mechanism.
 *
 *  @author Oscar Boykin
 *  @author Sam Ritchie
 */

object DRPCStore {
  def apply[Key, Value](nimbusHost: String, appID: String, port: Int = 3772)
  (implicit keyCodec: Bijection[Key, Array[Byte]], valCodec: Bijection[Value, Array[Byte]]) =
    new DRPCStore[Key,Value](nimbusHost, appID, port)
}

class DRPCStore[Key, Value](nimbusHost: String, appID: String, port: Int)
(implicit kBijection: Bijection[Key, Array[Byte]], vBijection: Bijection[Value, Array[Byte]])
extends ReadableStore[(Key, BatchID), Value] {
  val futurePool = FuturePool(Executors.newFixedThreadPool(4))

  val drpcClient = new DRPCClient(nimbusHost, port)

  implicit val pair2String: Bijection[(Key, BatchID), String] = RpcBijection.batchPair[Key]
  implicit val val2String: Bijection[Option[Value], String] = RpcBijection.option[Value]

  override def get(pair: (Key, BatchID)): Future[Option[Value]] =
    futurePool { drpcClient.execute(appID, pair.as[String]) }
      .map { _.as[Option[Value]] }
}

Not sure what the right async mongo driver to use is; possibly https://github.com/bwmcadams/hammersmith ?

(this may not be possible for multiGets where the keyset size can vary)

MemcacheLongMergeable extends Mergeable[String, Long]

Using the "incr" method:

https://github.com/twitter/finagle/blob/master/finagle-memcached/src/main/scala/com/twitter/finagle/memcached/Client.scala#L228

I'm thinking of a Hadoop output format for generating many Storehaus persistences.

The output format would:

• accept a number of shards and a shard function,
• assign a shard to each key,
• sort by (shard, key)
• creates N Stores on disk.

Paired with #47, and the VersionedTap in dfs-datastores, each output Store would be a proper VersionedStore.

This module should export the various scalacheck properties we've been using.

I started the effort in this feature branch:

https://github.com/twitter/storehaus/tree/feature/bdb

this looks complete, though it'll need tests.

If we do this, we can make the trait [-L,-Q, +X] which will make it somewhat easier to use in more cases.

cross can be moved to the companion object.

Some example code here:

https://github.com/azymnis/scalafish/blob/master/src/main/scala/org/zymnis/scalafish/zookeeper/Client.scala#L93

Might be good to use util-zk too:

https://github.com/twitter/util/tree/master/util-zk/src/main/scala/com/twitter/zk

I guess you have two stores here:

Data at nodes:
K = Path, V = (Data(Array[Byte]), Stat)

Children:
K = Path, V = (List[Path], Stat)

I'd love to see a port of Clojure core.cache's LIRS cache implementation over to scala:

https://github.com/clojure/core.cache/blob/master/src/main/clojure/clojure/core/cache.clj#L327

If a store could produce an iterator over its pairs it'd be possible to write a StorehausInputFormat and consume storehaus data from scalding.

trait IterableStore[-K, V] extends ReadableStore[K, V] with Iterable[(K, V)]

It may be convenient to have a built-in or explicit abstraction for stores where the read and write paths are distinct. Currently you can have a readable store but the next level of abstractions are Stores which readable and writable. It may be convenient to have a wrapping store that delegates all reads to a read-only store and all writes to a write-only store. I'm not sure what you would call this but it could just be a wrapper for both sides which delegate on puts/gets

// a better name maybe???
class PartionedStore(readStore: Store[...], writeStore: Store[...]) { 
  override def get(k: K) = readStore.get(k)
  override def put(kv: (k: K, v: Option[V]) = writeStore.put(kv)
  // yada yada
}

Related, it may be convenient to define an explicit write/append only store for things like Stats and logs.

If this is not worth while, close this issue out.

For network service backed stores that wish to do this, they can always define interfaces for supplying two underlying clients for store operations: one configured for reading and one configured for writing.

If we have K stores, we can return a value when K/2 + 1 have given the same result.

We can do the same for writable stores, except parallel writes, and then on read repair any missing/incorrect data.

Just a sketch:

ReadableStore seems well posed.

I have the idea of: SettableStore and UpdatableStore. This don't extend ReadableStore. Clearly, UpdatableStore extends SettableStore (just update and ignore the input). The SettableStore + ReadableStore can implement an UpdatableStore.

Lastly, I think we need to separate mutable and immutable. For instance, an LRU is a mutable ReadableStore (reading mutates it).

We can make a concurrent mutable version from the immutable one (just keep an internal var that we atomically update).

What do you think, Sam?

By default it could be EmptyReadableStore.

We could use a helper function like this to pad out KV store return values with Future.exception to conform to our interface:

val missing: Map[K, Future[Option[V]]] =
  (keys -- scalaM.keySet).map { k => (k, Future.exception(new MissingValueException("Missing value for key: " + k))) }.toMap

à la Dynamo or Cassandra. See 'Consistency' section here:
http://wiki.apache.org/cassandra/ArchitectureOverview

A LoggingStore is a ReadableStore that triggers a side-effect into a Mergeable on every call to get or multiGet:

class LoggingStore[-K, +V](store: ReadableStore[K, V], mergeable: Mergeable[K, Future[Option[V]]]) extends ReadableStore[K, V] {
  override def get(k: K) = {
    val ret = store.get(k)
    mergeable.merge(k -> ret)
    ret
  }
  override def multiGet[K1 <: K](ks: Set[K1]) = {
    val ret = store.multiGet(ks)
    mergeable.multiMerge(ret)
    ret
  }
  override def close { store.close }
}

pending some merges from the redis branch for resource cleanup and memcache mergables for new I'd like to put in some unit tests for the memcache module.

This method of construction would make it easy to, say, instantiate a TTLCache with an LRUCache, gaining the semantics of both caches.

I could be missing something but is CollectionOps#zipWith really worthy of its own utility? It's not a whole lot different from the std libs map + toMap

set.view.map(k => k -> lookup(k)).toMap

We need to encapsulate the text protocol of mysql into a MySqlValue typeclass which, I guess, is a special kind of Injection to String or perhaps Injection to ChannelBuffer (or both).

Then we could make the MySqlStore[K:MySqlValue, V: MySqlValue]

The wiki page on caches has some great info on write-through and write-back caches; this would allow a CachedStore that extends our CachedReadableStore.

Generally, we should study that page and make sure we have all the branches covered.

The recent addition of retry-able stores was nice but I think the way a client wants to retry should be a little more configurable.

Some interfaces where retry apis shine are

• twitter util zoo keeper's retry policy
• dispatch reboot's retries

I talked to @n8han, the author of dispatch, about extracting his interface which seems to be the most generic and he said it'd be cool. I hate to see this kind of thing be reinvented in multiple contexts so I created a standalone retry library based on the one in reboot.

The problems I face now are

1. the retry lib is based on scala's future interface, only available in 2.9.3 and 2.10+. Storehaus is published for 2.9.2 and 2.10 which means if I wanted to adapt this interface to storehaus I could only do it in modules published for 2.10.

2. The retry lib works with scala futures while storehaus is based on twttr futures. Scala's interface looks like it stole a lot from twitter so a bijection between the two should hopefully be the fix but I'd had to figure out a good way of handling the 2.9.2 case.

3. The retry lib needs to emulate sleeping asynchronously to simulate a pause between retries. Dispatch reboot uses netty's hashwheel timer to put a block of code on a timer. Twitter defines its own Timer interface. ( The zookeeper lib uses this ). In order to integrate a generic retry lib, a common retry interface would be required. This should not be hard. Potentially something like this should suffice.

4. Is storehaus open to non-twitter published dependencies? Since the retry library is outside the twitter lib lifecycle it could go in its own module storehaus. I publish everything to maven central so resolving should not be an issue.

We could use cassie which is built on finagle: https://github.com/twitter/cassie

Configure your cluster, keyspace, column family and pass it to the storehaus wrapper.

BufferingStore might accept a Timer as well and schedule a flush on some regular interval. This would require synchronizing access to the promise map and the wrapped store.

This should search readable stores in order for either

The first non-exceptional return value, or
The first non-exceptional, non-None return value

This should return Map[K,V] with empties filtered out. This brings back the old behavior.

Here's an idea:

trait ReadableStore[K,V] {
  def get(k: K): Future[Option[V]]
  def multiGet(ks: Set[K]): Future[Map[K, Future[Option[V]]]]
}

I tried to think of a better approach, and I this seems like the minimal change that can support combining stores in a way that doesn't result in all-or-nothing semantics.

https://github.com/twitter/storehaus/blob/develop/storehaus-core/src/main/scala/com/twitter/storehaus/ReadableStore.scala#L70

This would allow the user to use injections and bijections. flatMapValues could still make this version available.

Given a Seq[ReadableStore[K,V]], you can do a read and return the first non-None, non-exceptional value you get on get.

For multiGet, you can do the same for all the keys (keep waiting until you have all the results, or have exhausted the list.

This is similar to ReadableStore.select, but not quite the same.

Similarly, we send writes to all stores in a Seq[Store[_,K,V]].

We can model both of these as ReadableStore/Store and continue to combine with other combinators.

MemcacheStringMergeable extends Mergeable[String, String]

Using the "append" method:

https://github.com/twitter/finagle/blob/master/finagle-memcached/src/main/scala/com/twitter/finagle/memcached/Client.scala#L257

This would work for any other type with an append-like monoid.

Docs for implementing get, put and multiPut are pretty good:

https://github.com/fusesource/leveldbjni

Check out the LevelDB persistence implementation in ElephantDB for clues:

https://github.com/nathanmarz/elephantdb/blob/develop/elephantdb-leveldb/src/jvm/elephantdb/persistence/LevelDB.java

We should reimplement the old CommittingOnlineSink as a store combinator:

https://gist.github.com/sritchie/cb28fa90eedce4647cb6

Ideally, this store would build off of https://github.com/twitter/finagle/tree/master/finagle-mysql, but that's not currently published in maven central. @mariusaeriksen, any updates on the release plan for this module?

A colleague and myself curate a tumblr account for announcing release notes for scala projects using herald as the primary client. The recipe is simple, you drop a file named {version}.markdown in a notes directory under your source tree. Just run herald in your the directory containing the notes dir and herald will figure out the latest release and generate a preview and post to notes. This has been a great way for me to keep trace of what changes in my projects and document it for my users in a common location. Would this be something storehaus would be interested in doing? If you guys can pass along an email associated with a tumblr account. I can get you up and running.

Create a Storehaus wrapper over finagle-redis.

With the new multiGet return interface, we can use getsResult and capture information about hits, misses and failures.

http://twitter.github.com/finagle/api/finagle-memcached/com/twitter/finagle/memcached/Client.html