Elasticsearch 5.x singleton client with authentication #421

mars · 2017-08-14T20:05:09Z

Fixes both PIO-106 & PIO-114, replacing #372 & #420. These are combined because they each heavily revise the same class.

Authentication

Add optional username-password configuration for the new Elasticsearch 5 client; in pio-env.sh config:

# Optional basic HTTP auth
PIO_STORAGE_SOURCES_ELASTICSEARCH_USERNAME=my-name
PIO_STORAGE_SOURCES_ELASTICSEARCH_PASSWORD=my-secret

These credentials are sent in each Elasticsearch request as an HTTP Basic Authorization header.

Enables use of public-cloud, hosted Elasticsearch clusters, such as Bonsai on Heroku.

Singleton client

This PR moves to a singleton Elasticsearch RestClient which has built-in HTTP keep-alive and TCP connection pooling. Running on this branch, we've seen a 2x speed-up in predictions from the Universal Recommender with ES5, and the feared "cannot assign requested address" 😱 Elasticsearch connection errors have completely disappeared. Running pio batchpredict for 160K queries results in only 7 total TCP connections to Elasticsearch. Previously that would escalate to ~25,000 connections before denying further connections.

This fundamentally changes the interface for the new Elasticsearch 5.x REST client introduced with PredictionIO 0.11.0-incubating. With this changeset, the client is a single instance of org.elasticsearch.client.RestClient.

🚨 As a result of this change, any engine templates that directly use the Elasticsearch 5 StorageClient would require an update for compatibility. The change is this:

Original

val client: StorageClient = … // code to instantiate client
val restClient: RestClient = client.open()
try {
  restClient.performRequest(…)
} finally {
  restClient.close()
}

With this PR

val client: RestClient = … // code to instantiate client
client.performRequest(…)

No more balancing open & close as this is handled by using a new CleanupFunctions hook added to the framework in this PR.

Universal Recommender is the only template that I know of which directly uses the ES StorageClient outside of PredictionIO core. See example UR changes for compatibility with this PR.

Elasticsearch StorageClient changes

reimplemented as singleton
installs a cleanup function

See StorageClient

Core changes

A new CleanupFunctions hook has been added which enables developers of storage modules to register anonymous functions with CleanupFunctions.add { … } to be executed after Spark-related commands/workflows. The hook is called in a finally { CleanupFunctions.run() } from within:

pio import
pio export
pio train
pio batchpredict

Apologies for the huge indentation shifts from the requisite try-finally blocks:

try {
  // Freshly indented code.
} finally {
  CleanupFunctions.run()
}

…n pooling (PoolingNHttpClientConnectionManager)

…d gracefully.

…word)

…errors due to wrong version of class loaded

takezoe

@mars Sorry for slow response. I added some comments. Besides, some test patterns has been failed on Travis. Is it OK?

takezoe · 2017-08-22T06:59:23Z

core/src/main/scala/org/apache/predictionio/workflow/CleanupFunctions.scala

+  * be closed to allow the process to exit.
+  */
+object CleanupFunctions {
+  var functions: Seq[() => Unit] = Seq.empty[() => Unit]


This field should be private or protected at least.

takezoe · 2017-08-22T07:04:11Z

core/src/main/scala/org/apache/predictionio/workflow/CleanupFunctions.scala

+    * @param anonymous function containing cleanup code.
+    */
+  def run(): Unit = {
+    functions.map { case f => f() }


Use foreach instead of map if partial function does not have return value. Also case is not necessary in this case. As a result, you can write above code as:

functions.foreach { f => f() }

takezoe · 2017-08-22T07:09:06Z

data/src/main/scala/org/apache/predictionio/data/storage/Storage.scala

@@ -461,4 +463,11 @@ object Storage extends Logging {
      }.getOrElse(Map.empty)
    }
  )
+
+  def getStackTraceString(e: Throwable): String = {


You can get stacktrace from an exception using ExceptionUtils of commons-lang3.

import org.apache.commons.lang3.exception.ExceptionUtils val stackTrace = ExceptionUtils.getStackTrace(e)

takezoe · 2017-08-22T08:36:22Z

...search/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/StorageClient.scala

-case class ESClient(hosts: Seq[HttpHost]) {
-  def open(): RestClient = {
+object ESClient extends Logging {
+  var _sharedRestClient: Option[RestClient] = None


Is this client accessed from multi threads? If yes, I think adding volatile modifier to this field is necessary and open() and close() should be synchronized.

Thanks for the hints here. I suspected this would be an issue. I'm investigating how to make this threadsafe.

takezoe · 2017-08-22T08:39:24Z

...search/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/StorageClient.scala

+    if (!_sharedRestClient.isEmpty) {
+      _sharedRestClient.get.close()
+      _sharedRestClient = None
+    }


Typically, Option.get() is hated in the Scala world. You can write as follows instead:

_sharedRestClient.foreach { client => client.close() _sharedRestClient = None }

takezoe · 2017-08-22T08:47:08Z

...sticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/ESPEvents.scala

+          val jsonResponse = parse(EntityUtils.toString(response.getEntity))
+          val result = (jsonResponse \ "result").extract[String]
+          result match {
+            case "deleted" => true


true at here is no effect because the return type us Unit. I know it came from original code, but we should remove it.

takezoe · 2017-08-22T08:47:16Z

...sticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/ESPEvents.scala

+          result match {
+            case "deleted" => true
+            case _ =>
+              logger.error(s"[$result] Failed to update $index/$estype:$eventId")
              false


Same as above.

takezoe · 2017-08-22T08:47:23Z

...sticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/ESPEvents.scala

+        } catch {
+          case e: IOException =>
+            logger.error(s"Failed to update $index/$estype:$eventId", e)
+            false


Same as above.

mars · 2017-08-23T00:07:47Z

Cheers @takezoe I addressed all your Scala style & usage suggestions. Still need to take care of the threadsafety issue with the singleton client.

takezoe · 2017-08-23T07:25:39Z

I addressed all your Scala style & usage suggestions.

👍

Still need to take care of the threadsafety issue with the singleton client.

Yes, I'm reviewing the PIO codebase from this point of view now, but still my understanding about the whole PIO is not enough...

mars · 2017-08-23T15:24:47Z

Based on these Scala Concurrency/Thread Safety docs, I believe simply annotating @volatile will cause the synchronization needed for thread-safety in this case.

So, I updated this PR with that change.

mars · 2017-08-23T16:22:13Z

🏁

takezoe · 2017-08-29T04:56:23Z

@mars LGTM! but sorry for conflict with my previous commit: 6789dbe

mars · 2017-08-29T15:24:15Z

I will resolve these conflicts today and then merge this PR.

…on-with-auth

mars added 10 commits August 8, 2017 16:29

Migrate to singleton Elasticsearch client to use underlying connectio…

f30f27b

…n pooling (PoolingNHttpClientConnectionManager)

Log stacktrace for Storage initialization errors.

d999270

Remove shade rule that breaks Elasticsearch 5 client

dc4c31c

Collect & run cleanup functions to allow spark-submit processes to en…

7634a7a

…d gracefully.

Rename CleanupFunctions to match object name

5953451

Remove a missed client.close()

6c79e2f

Optional Elasticsearch 5 support for basic HTTP auth (username & pass…

bab0666

…word)

Beginning of unit tests for Elasticsearch REST client

073d734

Merge branch 'esclient-singleton' into esclient-auth

f7f74bd

Stabilize the sort order of CLASSPATH; fixes intermittent INSTANCE …

dc32c51

…errors due to wrong version of class loaded

This was referenced Aug 14, 2017

Elasticsearch basic HTTP authentication #372

Closed

[PIO-106] Elasticsearch 5.x StorageClient should reuse RestClient #420

Closed

mars changed the title ~~Elasticsearch singleton client with authentication~~ Elasticsearch 5x singleton client with authentication Aug 14, 2017

mars changed the title ~~Elasticsearch 5x singleton client with authentication~~ Elasticsearch 5.x singleton client with authentication Aug 14, 2017

Fix Scala style issues

438a122

mars mentioned this pull request Aug 14, 2017

Upgrade EsClient heroku/predictionio-engine-ur#5

Closed

Fix regression to put "plugins/*" in CLASSPATH

a479094

takezoe suggested changes Aug 22, 2017

View reviewed changes

Shade both ES and Apache HTTP (#1)

91baedc

mars mentioned this pull request Aug 22, 2017

EsClient upgrade (builtin) heroku/predictionio-engine-ur#7

Merged

improve Scala style & usage

2b18bbb

Thread-safety for CleanupFunctions

9089a9c

mars force-pushed the esclient-singleton-with-auth branch from 046d07c to 9089a9c Compare August 29, 2017 16:26

Merge remote-tracking branch 'upstream/develop' into esclient-singlet…

9f29059

…on-with-auth

Fix Elasticsearch REST client imports

9711174

asfgit closed this in bf84ede Aug 29, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Elasticsearch 5.x singleton client with authentication #421

Elasticsearch 5.x singleton client with authentication #421

mars commented Aug 14, 2017 •

edited

Loading

takezoe left a comment

takezoe Aug 22, 2017

takezoe Aug 22, 2017

takezoe Aug 22, 2017

takezoe Aug 22, 2017

mars Aug 23, 2017

takezoe Aug 22, 2017

takezoe Aug 22, 2017

takezoe Aug 22, 2017

takezoe Aug 22, 2017

mars commented Aug 23, 2017

takezoe commented Aug 23, 2017

mars commented Aug 23, 2017

mars commented Aug 23, 2017

takezoe commented Aug 29, 2017

mars commented Aug 29, 2017

Elasticsearch 5.x singleton client with authentication #421

Elasticsearch 5.x singleton client with authentication #421

Conversation

mars commented Aug 14, 2017 • edited Loading

Authentication

Singleton client

Original

With this PR

Elasticsearch StorageClient changes

Core changes

takezoe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mars commented Aug 23, 2017

takezoe commented Aug 23, 2017

mars commented Aug 23, 2017

mars commented Aug 23, 2017

takezoe commented Aug 29, 2017

mars commented Aug 29, 2017

mars commented Aug 14, 2017 •

edited

Loading