From cedeb98560066fd41a4d90feec4bc18d2588808e Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Wed, 7 Feb 2018 14:06:05 +0000 Subject: [PATCH 1/9] Added elastic4s dependencies. --- build.sbt | 6 +++++- project/Dependencies.scala | 8 ++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/build.sbt b/build.sbt index a67ca9b..fce883e 100644 --- a/build.sbt +++ b/build.sbt @@ -48,6 +48,10 @@ lazy val elasticsearch = project.in(file("instrumentation/elasticsearch")) .settings(CommonProject.settings) .settings( name := "logging-testkit-elasticsearch", - libraryDependencies += aws.logs, + libraryDependencies ++= Seq( + elastic4s.aws, + elastic4s.core, + elastic4s.http + ), coverageMinimum := 100 ) diff --git a/project/Dependencies.scala b/project/Dependencies.scala index 0ca037b..9953f47 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -30,6 +30,14 @@ object Dependencies { val reactive: ModuleID = "io.monix" %% "monix-reactive" % version } + object elastic4s { + private val version = "6.1.4" + + val aws: ModuleID = "com.sksamuel.elastic4s" %% "elastic4s-aws" % version + val core: ModuleID = "com.sksamuel.elastic4s" %% "elastic4s-core" % version + val http: ModuleID = "com.sksamuel.elastic4s" %% "elastic4s-http" % version + } + val scalacheck: ModuleID = "org.scalacheck" %% "scalacheck" % "1.13.5" val scalatest: ModuleID = "org.scalatest" %% "scalatest" % "3.0.5" } From eb521392b7ea9d99c8682f54070a179ba6b976cb Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Wed, 7 Feb 2018 14:14:21 +0000 Subject: [PATCH 2/9] Implemented ElasticSearchLogSource. --- .../ElasticSearchLogLineSource.scala | 27 -------- .../ElasticSearchLogSource.scala | 66 +++++++++++++++++++ 2 files changed, 66 insertions(+), 27 deletions(-) delete mode 100644 instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogLineSource.scala create mode 100644 instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala diff --git a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogLineSource.scala b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogLineSource.scala deleted file mode 100644 index d736a03..0000000 --- a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogLineSource.scala +++ /dev/null @@ -1,27 +0,0 @@ -// Copyright 2018 Cake Solutions Limited - -package net.cakesolutions.testkit.logging.elasticsearch - -import scala.concurrent.Promise - -import io.circe.Json -import monix.execution.Scheduler -import monix.reactive.observers.Subscriber -import net.cakesolutions.testkit.logging.LineLoggingSource - -// $COVERAGE-OFF$ disabled until this is implemented - -object ElasticSearchLogLineSource extends LineLoggingSource[Json] { - - /** @see net.cakesolutions.testkit.logging.LoggingSource */ - override protected def subscriberPolling( - subscriber: Subscriber[Json], - cancelP: Promise[Unit] - )(implicit - scheduler: Scheduler - ): Unit = { - ??? // TODO: needs implementing - } -} - -// $COVERAGE-ON$ diff --git a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala new file mode 100644 index 0000000..50a284e --- /dev/null +++ b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala @@ -0,0 +1,66 @@ +// Copyright 2018 Cake Solutions Limited + +package net.cakesolutions.testkit.logging.elasticsearch + +import scala.concurrent.Future +import com.sksamuel.elastic4s.aws.{Aws4ElasticClient, Aws4ElasticConfig} +import com.sksamuel.elastic4s.http.RequestFailure +import com.sksamuel.elastic4s.http.ElasticDsl._ +import com.sksamuel.elastic4s.searches.SearchDefinition +import com.typesafe.scalalogging.Logger +import monix.execution.{Cancelable, Scheduler} +import monix.reactive.{Observable, OverflowStrategy} +import net.cakesolutions.testkit.config.Configuration.Logging + +import scala.util.control.NonFatal + +final class ElasticSearchLogSource( + search: SearchDefinition, + config: Aws4ElasticConfig +) { + import ElasticSearchLogSource._ + + private val logger = Logger(Logging.name) + + def source()(implicit scheduler: Scheduler): Observable[String] = + Observable.create[String]( + OverflowStrategy.Unbounded + ) { subscriber => + try { + scheduler.execute { + new Runnable { + override def run(): Unit = { + val awsElasticClient = Aws4ElasticClient.apply(config) + Observable + .fromFuture(awsElasticClient.execute(search).flatMap { + case Left(failure) => + Future.failed( + ElasticSearchRequestFailureException(failure) + ) + case Right(result) => + Future.successful(result) + }) + .flatMap { result => + Observable.fromIterable( + result.result.hits.hits.map(_.sourceAsString) + ) + } + } + } + } + } catch { + case NonFatal(error) => + logger.error("Log parsing error", error) + subscriber.onError(error) + } + Cancelable(() => subscriber.onComplete()) + } +} + +object ElasticSearchLogSource { + case class ElasticSearchRequestFailureException( + requestFailure: RequestFailure + ) extends Exception( + s"ElasticSearch request failed: ${requestFailure.error.reason}" + ) +} From e57c6d78f0f0f751b1f746d58df4a0d967257f4f Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Thu, 8 Feb 2018 10:21:02 +0000 Subject: [PATCH 3/9] Remove unnecessary wrapper code. --- .../ElasticSearchLogSource.scala | 57 +++++++------------ 1 file changed, 20 insertions(+), 37 deletions(-) diff --git a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala index 50a284e..e7e5327 100644 --- a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala +++ b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala @@ -8,53 +8,36 @@ import com.sksamuel.elastic4s.http.RequestFailure import com.sksamuel.elastic4s.http.ElasticDsl._ import com.sksamuel.elastic4s.searches.SearchDefinition import com.typesafe.scalalogging.Logger -import monix.execution.{Cancelable, Scheduler} -import monix.reactive.{Observable, OverflowStrategy} +import monix.execution.Scheduler +import monix.reactive.Observable import net.cakesolutions.testkit.config.Configuration.Logging -import scala.util.control.NonFatal - final class ElasticSearchLogSource( - search: SearchDefinition, + searchDef: SearchDefinition, config: Aws4ElasticConfig ) { + import ElasticSearchLogSource._ private val logger = Logger(Logging.name) - def source()(implicit scheduler: Scheduler): Observable[String] = - Observable.create[String]( - OverflowStrategy.Unbounded - ) { subscriber => - try { - scheduler.execute { - new Runnable { - override def run(): Unit = { - val awsElasticClient = Aws4ElasticClient.apply(config) - Observable - .fromFuture(awsElasticClient.execute(search).flatMap { - case Left(failure) => - Future.failed( - ElasticSearchRequestFailureException(failure) - ) - case Right(result) => - Future.successful(result) - }) - .flatMap { result => - Observable.fromIterable( - result.result.hits.hits.map(_.sourceAsString) - ) - } - } - } - } - } catch { - case NonFatal(error) => - logger.error("Log parsing error", error) - subscriber.onError(error) + def source()(implicit scheduler: Scheduler): Observable[String] = { + val awsElasticClient = Aws4ElasticClient.apply(config) + Observable + .fromFuture(awsElasticClient.execute(searchDef).flatMap { + case Left(failure) => + Future.failed( + ElasticSearchRequestFailureException(failure) + ) + case Right(result) => + Future.successful(result) + }) + .flatMap { result => + Observable.fromIterable( + result.result.hits.hits.map(_.sourceAsString) + ) } - Cancelable(() => subscriber.onComplete()) - } + } } object ElasticSearchLogSource { From 65174e7fe23380cdb11d2e609c392465541d9e0c Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Thu, 8 Feb 2018 10:44:54 +0000 Subject: [PATCH 4/9] Removed unnecessary apply. --- .../testkit/logging/elasticsearch/ElasticSearchLogSource.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala index e7e5327..19e24ad 100644 --- a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala +++ b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala @@ -22,7 +22,7 @@ final class ElasticSearchLogSource( private val logger = Logger(Logging.name) def source()(implicit scheduler: Scheduler): Observable[String] = { - val awsElasticClient = Aws4ElasticClient.apply(config) + val awsElasticClient = Aws4ElasticClient(config) Observable .fromFuture(awsElasticClient.execute(searchDef).flatMap { case Left(failure) => From a1c783e2440c79978ab1a074fe9758de23100e65 Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Thu, 8 Feb 2018 10:53:43 +0000 Subject: [PATCH 5/9] Formatting cleanup. --- .../elasticsearch/ElasticSearchLogSource.scala | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala index 19e24ad..9f25373 100644 --- a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala +++ b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala @@ -16,7 +16,6 @@ final class ElasticSearchLogSource( searchDef: SearchDefinition, config: Aws4ElasticConfig ) { - import ElasticSearchLogSource._ private val logger = Logger(Logging.name) @@ -24,14 +23,14 @@ final class ElasticSearchLogSource( def source()(implicit scheduler: Scheduler): Observable[String] = { val awsElasticClient = Aws4ElasticClient(config) Observable - .fromFuture(awsElasticClient.execute(searchDef).flatMap { - case Left(failure) => - Future.failed( - ElasticSearchRequestFailureException(failure) - ) - case Right(result) => - Future.successful(result) - }) + .fromFuture( + awsElasticClient.execute(searchDef).flatMap { + case Left(failure) => + Future.failed(ElasticSearchRequestFailureException(failure)) + case Right(result) => + Future.successful(result) + } + ) .flatMap { result => Observable.fromIterable( result.result.hits.hits.map(_.sourceAsString) From 159d87826e62a3fb660e6ae610a9ad7e83b79f7b Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Thu, 8 Feb 2018 10:57:40 +0000 Subject: [PATCH 6/9] Disable unused import warnings in console. --- project/CommonProject.scala | 2 ++ 1 file changed, 2 insertions(+) diff --git a/project/CommonProject.scala b/project/CommonProject.scala index 5d29c7d..74cf0a8 100644 --- a/project/CommonProject.scala +++ b/project/CommonProject.scala @@ -31,6 +31,8 @@ object CommonProject { "-Ypartial-unification", "-Xfatal-warnings" ), + // Disable unused import warnings in Scala console. + scalacOptions in (Compile, console) -= "-Ywarn-unused-import", scalacOptions in (Compile, doc) ++= { val nm = (name in(Compile, doc)).value val ver = (version in(Compile, doc)).value From 662c490891325bc79980b25051b29de278a6fceb Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Thu, 8 Feb 2018 11:06:57 +0000 Subject: [PATCH 7/9] Pass search as parameter to Observable creation method. --- ...urce.scala => ElasticSearchLogClient.scala} | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) rename instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/{ElasticSearchLogSource.scala => ElasticSearchLogClient.scala} (76%) diff --git a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala similarity index 76% rename from instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala rename to instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala index 9f25373..3905c08 100644 --- a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogSource.scala +++ b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala @@ -12,19 +12,19 @@ import monix.execution.Scheduler import monix.reactive.Observable import net.cakesolutions.testkit.config.Configuration.Logging -final class ElasticSearchLogSource( - searchDef: SearchDefinition, - config: Aws4ElasticConfig -) { - import ElasticSearchLogSource._ +final class ElasticSearchLogClient(config: Aws4ElasticConfig) { + import ElasticSearchLogClient._ private val logger = Logger(Logging.name) - def source()(implicit scheduler: Scheduler): Observable[String] = { - val awsElasticClient = Aws4ElasticClient(config) + val elasticSearchClient = Aws4ElasticClient(config) + + def search(searchDef: SearchDefinition)( + implicit scheduler: Scheduler + ): Observable[String] = { Observable .fromFuture( - awsElasticClient.execute(searchDef).flatMap { + elasticSearchClient.execute(searchDef).flatMap { case Left(failure) => Future.failed(ElasticSearchRequestFailureException(failure)) case Right(result) => @@ -39,7 +39,7 @@ final class ElasticSearchLogSource( } } -object ElasticSearchLogSource { +object ElasticSearchLogClient { case class ElasticSearchRequestFailureException( requestFailure: RequestFailure ) extends Exception( From ae991b1f40d504ffa2eb3b24009b88e91993f748 Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Thu, 8 Feb 2018 13:24:09 +0000 Subject: [PATCH 8/9] Remove unused logger. --- .../logging/elasticsearch/ElasticSearchLogClient.scala | 4 ---- 1 file changed, 4 deletions(-) diff --git a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala index 3905c08..135877d 100644 --- a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala +++ b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala @@ -7,16 +7,12 @@ import com.sksamuel.elastic4s.aws.{Aws4ElasticClient, Aws4ElasticConfig} import com.sksamuel.elastic4s.http.RequestFailure import com.sksamuel.elastic4s.http.ElasticDsl._ import com.sksamuel.elastic4s.searches.SearchDefinition -import com.typesafe.scalalogging.Logger import monix.execution.Scheduler import monix.reactive.Observable -import net.cakesolutions.testkit.config.Configuration.Logging final class ElasticSearchLogClient(config: Aws4ElasticConfig) { import ElasticSearchLogClient._ - private val logger = Logger(Logging.name) - val elasticSearchClient = Aws4ElasticClient(config) def search(searchDef: SearchDefinition)( From e21babae363afcf9e1d56f68fd65d03ff8ce7937 Mon Sep 17 00:00:00 2001 From: Alex Horsman Date: Thu, 8 Feb 2018 13:54:20 +0000 Subject: [PATCH 9/9] Added ScalaDocs. --- .../ElasticSearchLogClient.scala | 22 +++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala index 135877d..af86afe 100644 --- a/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala +++ b/instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogClient.scala @@ -10,11 +10,23 @@ import com.sksamuel.elastic4s.searches.SearchDefinition import monix.execution.Scheduler import monix.reactive.Observable -final class ElasticSearchLogClient(config: Aws4ElasticConfig) { +/** + * Client for querying logs from ElasticSearch. + * + * @param config AWS elasticsearch configuration + */ +class ElasticSearchLogClient(config: Aws4ElasticConfig) { import ElasticSearchLogClient._ - val elasticSearchClient = Aws4ElasticClient(config) + private val elasticSearchClient = Aws4ElasticClient(config) + /** + * Query ElasticSearch and return the results as an Observable. + * + * @param searchDef search query + * @param scheduler monix scheduler + * @return query results as observable stream + */ def search(searchDef: SearchDefinition)( implicit scheduler: Scheduler ): Observable[String] = { @@ -36,6 +48,12 @@ final class ElasticSearchLogClient(config: Aws4ElasticConfig) { } object ElasticSearchLogClient { + + /** + * Execution of an ElasticSearch request failed. + * + * @param requestFailure failure response + */ case class ElasticSearchRequestFailureException( requestFailure: RequestFailure ) extends Exception(