Reporting store algebra's

storehaus-algebra/src/main/scala/com/twitter/storehaus/algebra/reporting/Reporter.scala

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2013 Twitter Inc.
+ *
+ * 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 com.twitter.storehaus.algebra.reporting
+
+import com.twitter.storehaus.{Store, ReadableStore, WritableStore}
+import com.twitter.storehaus.algebra.Mergeable
+import com.twitter.util.{Future, Time}
+
+
+object Reporter {
+  def sideEffect[T, U](params: U, f: Future[T], sideEffect: (U, Future[T]) => Future[Unit]): Future[T] =
+        Future.join(f, sideEffect(params, f)).map(_._1)
+
+  def sideEffect[T, K, U](params: U, f: Map[K, Future[T]], sideEffect: (U, Map[K, Future[T]]) => Map[K, Future[Unit]]): Map[K, Future[T]] = {
+        val effected = sideEffect(params, f)
+        f.map{case (k, v) =>
+            val unitF = effected.getOrElse(k, sys.error("Reporter for multi side effect didn't return a future for key" + k.toString))
+            (k, Future.join(v, unitF).map(_._1))
+        }
+      }
+}
+
+trait ReadableStoreReporter[S <: ReadableStore[K, V], K, V] extends ReadableStore[K, V] {
+  def self: S
+  override def get(k: K): Future[Option[V]] = Reporter.sideEffect(k, self.get(k), traceGet)
+  override def multiGet[K1 <: K](keys: Set[K1]) = Reporter.sideEffect(keys, self.multiGet(keys), traceMultiGet)
+
+  protected def traceMultiGet[K1 <: K](ks: Set[K1], request: Map[K1, Future[Option[V]]]): Map[K1, Future[Unit]]
+  protected def traceGet(k: K, request: Future[Option[V]]): Future[Unit]
+  override def close(time: Time) = self.close(time)
+}
+
+
+trait WritableStoreReporter[S <: WritableStore[K, V],K, V] extends WritableStore[K, V] {
+  def self: S
+  override def put(kv: (K, V)): Future[Unit] = Reporter.sideEffect(kv, self.put(kv), tracePut)
+  override def multiPut[K1 <: K](kvs: Map[K1, V]) = Reporter.sideEffect(kvs, self.multiPut(kvs), traceMultiPut)
+
+  protected def tracePut(kv: (K, V), request: Future[Unit]): Future[Unit]
+  protected def traceMultiPut[K1 <: K](kvs: Map[K1, V], request: Map[K1, Future[Unit]]): Map[K1, Future[Unit]]
+  override def close(time: Time) = self.close(time)
+}
+
+
+trait MergeableReporter[S <: Mergeable[K, V],K, V] extends Mergeable[K, V] {
+  def self: S
+  override def merge(kv: (K, V)) = Reporter.sideEffect(kv, self.merge(kv), traceMerge)
+  override def multiMerge[K1 <: K](kvs: Map[K1, V]) = Reporter.sideEffect(kvs, self.multiMerge(kvs), traceMultiMerge)
+
+  protected def traceMerge(kv: (K, V), request: Future[Option[V]]): Future[Unit]
+  protected def traceMultiMerge[K1 <: K](kvs: Map[K1, V], request: Map[K1, Future[Option[V]]]): Map[K1, Future[Unit]]
+  override def close(time: Time) = self.close(time)
+}
+
+
+trait StoreReporter[S <: Store[K, V], K, V] extends ReadableStoreReporter[S, K, V] with WritableStoreReporter[S, K, Option[V]] {
+  def self: S
+}

storehaus-algebra/src/test/scala/com/twitter/storehaus/algebra/reporting/ReportingMergeableStoreProperties.scala

@@ -0,0 +1,98 @@
+
+/*
+ * Copyright 2013 Twitter Inc.
+ *
+ * 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 com.twitter.storehaus.algebra.reporting
+
+import com.twitter.util.{ Await, Future }
+import org.scalacheck.{ Arbitrary, Properties }
+import org.scalacheck.Gen.choose
+import org.scalacheck.Prop._
+
+import com.twitter.storehaus.algebra._
+
+object ReportingMergeableStoreProperties extends Properties("ReportingMergeableStore") {
+  import MergeableStoreProperties.{mergeableStoreTest, newStore}
+
+
+
+  class DummyReporter[K, V](val self: Mergeable[K, V]) extends MergeableProxy[K, V] with MergeableReporter[Mergeable[K, V], K, V] {
+    def traceMerge(kv: (K, V), request: Future[Option[V]]) = request.unit
+    def traceMultiMerge[K1 <: K](kvs: Map[K1, V], request: Map[K1, Future[Option[V]]]) = request.mapValues(_.unit)
+  }
+
+  property("Mergable stat store obeys the mergeable store proporites") =
+    mergeableStoreTest {
+      val store = newStore[Int, Int]
+      new MergeableStoreProxy[Int, Int] with MergeableReporter[Mergeable[Int, Int], Int, Int] {
+        val self = store
+        def traceMerge(kv: (Int, Int), request: Future[Option[Int]]) = request.unit
+        def traceMultiMerge[K1 <: Int](kvs: Map[K1, Int], request: Map[K1, Future[Option[Int]]]) = request.mapValues(_.unit)
+      }
+    }
+
+  property("merge Some/None count matches") = forAll { (base: Map[Int, Int], merge: Map[Int, Int]) =>
+        var mergeWithSomeCount = 0
+        var mergeWithNoneCount = 0
+        val baseStore = MergeableStore.fromStore(newStore[Int, Int])
+        baseStore.multiMerge(base)
+
+        val wrappedStore = new DummyReporter[Int, Int](baseStore) {
+          override def traceMerge(kv: (Int, Int), request: Future[Option[Int]]) = {
+            request.map { optV =>
+              optV match {
+                case Some(_) => mergeWithSomeCount += 1
+                case None => mergeWithNoneCount += 1
+                }
+            }.unit
+          }
+        }
+
+        merge.map(kv => wrappedStore.merge((kv._1, kv._2)))
+
+        val existsBeforeList = merge.keySet.toList.map(k => base.get(k))
+
+        existsBeforeList.collect{case Some(_) => 1}.size == mergeWithSomeCount &&
+          existsBeforeList.collect{case None => 1}.size == mergeWithNoneCount
+  }
+
+  property("multiMerge Some/None count matches") = forAll { (base: Map[Int, Int], merge: Map[Int, Int]) =>
+        var mergeWithSomeCount = 0
+        var mergeWithNoneCount = 0
+        val baseStore = MergeableStore.fromStore(newStore[Int, Int])
+        baseStore.multiMerge(base)
+
+        val wrappedStore = new DummyReporter[Int, Int](baseStore) {
+          override def traceMultiMerge[K1 <: Int](kvs: Map[K1, Int], request: Map[K1, Future[Option[Int]]]) = {
+            request.mapValues{optV =>
+              optV.map{ v =>
+                v match {
+                  case Some(_) => mergeWithSomeCount += 1
+                  case None => mergeWithNoneCount += 1
+                }
+              }.unit
+            }
+          }
+        }
+
+        wrappedStore.multiMerge(merge)
+
+        val existsBeforeList = merge.keySet.toList.map(k => base.get(k))
+
+        existsBeforeList.collect{case Some(_) => 1}.size == mergeWithSomeCount &&
+          existsBeforeList.collect{case None => 1}.size == mergeWithNoneCount
+  }
+}

storehaus-algebra/src/test/scala/com/twitter/storehaus/algebra/reporting/ReportingReadableStoreProperties.scala

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2013 Twitter Inc.
+ *
+ * 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 com.twitter.storehaus.algebra.reporting
+
+import com.twitter.util.{ Await, Future }
+import com.twitter.storehaus.{ReadableStore, ReadableStoreProxy}
+import org.scalacheck.{ Arbitrary, Properties }
+import org.scalacheck.Gen.choose
+import org.scalacheck.Prop._
+
+object ReportingReadableStoreProperties extends Properties("ReportingReadableStore") {
+    /**
+    * get returns none when not in either store
+    */
+
+   class DummyReporter[K, V](val self: ReadableStore[K, V]) extends ReadableStoreProxy[K, V] with ReadableStoreReporter[ReadableStore[K, V], K, V] {
+    def traceMultiGet[K1 <: K](ks: Set[K1], request: Map[K1, Future[Option[V]]]) = request.mapValues(_.unit)
+    def traceGet(k: K, request: Future[Option[V]]) = request.unit
+  }
+
+  def buildStoreRunQueries[K, V](mA: Map[K, V], others: Set[K], builder: (ReadableStore[K, V]) => ReadableStore[K, V]) = {
+    val baseStore = ReadableStore.fromMap(mA)
+    val wrappedStore = builder(baseStore)
+    val expanded: Set[K] = (mA.keySet ++ others)
+    // We use call to list, or it keeps the results of the map as a set and we loose data
+    expanded.toList.map{k: K => (mA.get(k), Await.result(wrappedStore.get(k)))}
+  }
+
+  def buildStoreRunMultiGetQueries[K, V](mA: Map[K, V], others: Set[K], builder: (ReadableStore[K, V]) => ReadableStore[K, V]) = {
+    val baseStore = ReadableStore.fromMap(mA)
+    val wrappedStore = builder(baseStore)
+    val expanded: Set[K] = (mA.keySet ++ others)
+    // We use call to list, or it keeps the results of the map as a set and we loose data
+    (expanded.map{k => (k, mA.get(k))}.toMap,
+      wrappedStore.multiGet(expanded).map{case (k, futureV) => (k, Await.result(futureV))})
+  }
+
+  property("Stats store matches raw get for all queries") = forAll { (mA: Map[Int, String], others: Set[Int]) =>
+        def reporter(store: ReadableStore[Int, String]) = new DummyReporter[Int, String](store)
+        val queryResults = buildStoreRunQueries(mA, others, reporter)
+        queryResults.forall{case (a, b) => a == b}
+  }
+
+  property("Present/Absent count matches") = forAll { (mA: Map[Int, String], others: Set[Int]) =>
+        var presentCount = 0
+        var absentCount = 0
+        def reporter(store: ReadableStore[Int, String]) = new DummyReporter[Int, String](store) {
+          override def traceGet(k: Int, request: Future[Option[String]]) = {
+            request.map{ optV =>
+              optV match {
+                case Some(_) => presentCount += 1
+                case None => absentCount += 1
+              }
+            }
+          }
+        }
+        val queryResults = buildStoreRunQueries(mA, others, reporter)
+        val wrappedResults = queryResults.map(_._2)
+        val referenceResults = queryResults.map(_._2)
+        wrappedResults.collect{case Some(b) => b}.size == presentCount &&
+          wrappedResults.collect{case None => 1}.size == absentCount
+  }
+
+  property("Stats store matches raw get for multiget all queries") = forAll { (mA: Map[Int, String], others: Set[Int]) =>
+        def reporter(store: ReadableStore[Int, String]) = new DummyReporter[Int, String](store)
+        val (mapRes, storeResults) = buildStoreRunMultiGetQueries(mA, others, reporter)
+        mapRes.size == storeResults.size &&
+          mapRes.keySet.forall(k => mapRes.get(k) == storeResults.get(k))
+  }
+
+  property("Present/Absent count matches in multiget") = forAll { (mA: Map[Int, String], others: Set[Int]) =>
+        var presentCount = 0
+        var absentCount = 0
+
+        def reporter(store: ReadableStore[Int, String]) = new DummyReporter[Int, String](store) {
+          override def traceMultiGet[K1 <: Int](ks: Set[K1], request: Map[K1, Future[Option[String]]]) = {
+            request.mapValues{fOptV =>
+              fOptV.map {optV =>
+                optV match {
+                  case Some(_) => presentCount += 1
+                  case None => absentCount += 1
+                }
+              }
+            }
+          }
+        }
+
+        val (_, storeResults) = buildStoreRunMultiGetQueries(mA, others, reporter)
+        storeResults.values.collect{case Some(b) => b}.size == presentCount &&
+          storeResults.values.collect{case None => 1}.size == absentCount
+  }
+}

storehaus-algebra/src/test/scala/com/twitter/storehaus/algebra/reporting/ReportingStoreProperties.scala

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2013 Twitter Inc.
+ *
+ * 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 com.twitter.storehaus.algebra.reporting
+
+import com.twitter.util.{ Await, Future }
+import org.scalacheck.{ Arbitrary, Properties }
+import org.scalacheck.Gen.choose
+import org.scalacheck.Prop._
+
+import com.twitter.storehaus._
+import com.twitter.storehaus.algebra._
+
+object ReportingStoreProperties extends Properties("ReportingStore") {
+  def newStore[K, V] = new JMapStore[K, V]
+
+  class DummyReporter[K, V](val self: Store[K, V]) extends StoreProxy[K, V] with StoreReporter[Store[K, V], K, V] {
+    def traceMultiGet[K1 <: K](ks: Set[K1], request: Map[K1, Future[Option[V]]]) = request.mapValues(_.unit)
+    def traceGet(k: K, request: Future[Option[V]]) = request.unit
+
+    def tracePut(kv: (K, Option[V]), request: Future[Unit]) = request.unit
+    def traceMultiPut[K1 <: K](kvs: Map[K1, Option[V]], request: Map[K1, Future[Unit]]) = request.mapValues(_.unit)
+  }
+
+  property("Put Some/None count matches") = forAll { (inserts: Map[Int, Option[Int]]) =>
+        var putSomeCount = 0
+        var putNoneCount = 0
+        val baseStore = newStore[Int, Int]
+
+        val wrappedStore = new DummyReporter[Int, Int](baseStore) {
+          override def tracePut(kv: (Int, Option[Int]), request: Future[Unit]) = {
+            Future {
+                kv._2 match {
+                case Some(_) => putSomeCount += 1
+                case None => putNoneCount += 1
+                }
+            }.unit
+          }
+        }
+
+        inserts.foreach{ i =>
+          wrappedStore.put(i._1, i._2)
+        }
+        inserts.map(_._2).collect{case Some(b) => b}.size == putSomeCount &&
+          inserts.map(_._2).collect{case None => 1}.size == putNoneCount
+  }
+
+  property("MultiPut Some/None count matches") = forAll { (inserts: Map[Int, Option[Int]]) =>
+        var multiPutSomeCount = 0
+        var multiPutNoneCount = 0
+        val baseStore = newStore[Int, Int]
+
+        val wrappedStore = new DummyReporter[Int, Int](baseStore) {
+         override def traceMultiPut[K1 <: Int](kvs: Map[K1, Option[Int]], request: Map[K1, Future[Unit]]): Map[K1, Future[Unit]] = {
+            kvs.mapValues {v =>
+              Future {
+                v match {
+                case Some(_) => multiPutSomeCount += 1
+                case None => multiPutNoneCount += 1
+                }
+              }.unit
+            }
+          }
+        }
+
+        wrappedStore.multiPut(inserts)
+
+        inserts.map(_._2).collect{case Some(b) => b}.size == multiPutSomeCount &&
+          inserts.map(_._2).collect{case None => 1}.size == multiPutNoneCount
+  }
+}