You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@predictionio.apache.org by mars <gi...@git.apache.org> on 2017/08/14 20:05:12 UTC
[GitHub] incubator-predictionio pull request #421: Elasticsearch singleton client wit...
GitHub user mars opened a pull request:
https://github.com/apache/incubator-predictionio/pull/421
Elasticsearch singleton client with authentication
Fixes both [PIO-106](https://issues.apache.org/jira/browse/PIO-106) & [PIO-114](https://issues.apache.org/jira/browse/PIO-114), replacing https://github.com/apache/incubator-predictionio/pull/372. 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:
```bash
# 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](https://elements.heroku.com/addons/bonsai).
## 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](https://github.com/apache/incubator-predictionio/tree/develop/storage/elasticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch)** introduced with PredictionIO 0.11.0-incubating. With this changeset, the `client` is a single instance of [`org.elasticsearch.client.RestClient`](https://github.com/elastic/elasticsearch/blob/master/client/rest/src/main/java/org/elasticsearch/client/RestClient.java).
🚨 **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
```scala
val client: StorageClient = … // code to instantiate client
val restClient: RestClient = client.open()
try {
restClient.performRequest(…)
} finally {
restClient.close()
}
```
### With this PR
```scala
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](https://github.com/actionml/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](https://github.com/heroku/predictionio-engine-ur/compare/esclient-singleton).
### Elasticsearch StorageClient changes
* reimplemented as singleton
* installs a cleanup function
See [StorageClient](https://github.com/apache/incubator-predictionio/compare/develop...mars:esclient-singleton?expand=1#diff-2926f4cfd93ccb02320e2a9503ccd223)
### Core changes
A new [`CleanupFunctions`](https://github.com/apache/incubator-predictionio/compare/develop...mars:esclient-singleton?expand=1#diff-2a958821ac58f019fbce38540c775f19) 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:
```scala
try {
// Freshly indented code.
} finally {
CleanupFunctions.run()
}
```
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mars/incubator-predictionio esclient-singleton-with-auth
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-predictionio/pull/421.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #421
----
commit f30f27bcc09a397efb42a7923938beceaeac37bf
Author: Mars Hall <ma...@heroku.com>
Date: 2017-08-08T23:29:15Z
Migrate to singleton Elasticsearch client to use underlying connection pooling (PoolingNHttpClientConnectionManager)
commit d99927089a41cb85f525cb74bdf394eed4686bf2
Author: Mars Hall <ma...@heroku.com>
Date: 2017-08-10T03:00:58Z
Log stacktrace for Storage initialization errors.
commit dc4c31cbcddbb3b281d52b8099e210adc546d1ed
Author: Mars Hall <ma...@heroku.com>
Date: 2017-08-10T22:55:38Z
Remove shade rule that breaks Elasticsearch 5 client
commit 7634a7ab720239d5f8efda85f67b26bdaff797f8
Author: Mars Hall <ma...@heroku.com>
Date: 2017-08-10T22:59:01Z
Collect & run cleanup functions to allow spark-submit processes to end gracefully.
commit 5953451f40e554eafa887328122c794edbbd8f1d
Author: Mars Hall <ma...@heroku.com>
Date: 2017-08-11T00:06:24Z
Rename CleanupFunctions to match object name
commit 6c79e2f30e166185e8a2afbda27689b387a6b6d1
Author: Mars Hall <ma...@heroku.com>
Date: 2017-08-11T18:26:12Z
Remove a missed `client.close()`
commit bab0666f24af19228626aae4c0be39ca890a1f6a
Author: Mars Hall <ma...@heroku.com>
Date: 2017-04-19T18:37:18Z
Optional Elasticsearch 5 support for basic HTTP auth (username & password)
commit 073d734a9efe321314964b48764a27b392610761
Author: Mars Hall <ma...@heroku.com>
Date: 2017-04-22T01:28:46Z
Beginning of unit tests for Elasticsearch REST client
commit f7f74bd87fa158168740715e1f42eaa57c3cf0a9
Author: Mars Hall <ma...@heroku.com>
Date: 2017-08-11T19:04:24Z
Merge branch 'esclient-singleton' into esclient-auth
commit dc32c51e76cc7c358f59ede7c9f42a8f2fb7fe45
Author: Mars Hall <ma...@heroku.com>
Date: 2017-05-23T20:20:03Z
Stabilize the sort order of CLASSPATH; fixes intermittent `INSTANCE` errors due to wrong version of class loaded
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio issue #421: Elasticsearch 5.x singleton client with a...
Posted by mars <gi...@git.apache.org>.
Github user mars commented on the issue:
https://github.com/apache/incubator-predictionio/pull/421
🏁
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134419425
--- Diff: storage/elasticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/ESPEvents.scala ---
@@ -107,37 +107,32 @@ class ESPEvents(client: ESClient, config: StorageClientConfig, index: String)
eventIds: RDD[String],
appId: Int, channelId: Option[Int])(sc: SparkContext): Unit = {
val estype = getEsType(appId, channelId)
- val restClient = client.open()
- try {
- eventIds.foreachPartition { iter =>
- iter.foreach { eventId =>
- try {
- val json =
- ("query" ->
- ("term" ->
- ("eventId" -> eventId)))
- val entity = new NStringEntity(compact(render(json)), ContentType.APPLICATION_JSON)
- val response = restClient.performRequest(
- "POST",
- s"/$index/$estype/_delete_by_query",
- Map("refresh" -> ESUtils.getEventDataRefresh(config)).asJava)
- val jsonResponse = parse(EntityUtils.toString(response.getEntity))
- val result = (jsonResponse \ "result").extract[String]
- result match {
- case "deleted" => true
- case _ =>
- logger.error(s"[$result] Failed to update $index/$estype:$eventId")
- false
- }
- } catch {
- case e: IOException =>
- logger.error(s"Failed to update $index/$estype:$eventId", e)
+ eventIds.foreachPartition { iter =>
+ iter.foreach { eventId =>
+ try {
+ val json =
+ ("query" ->
+ ("term" ->
+ ("eventId" -> eventId)))
+ val entity = new NStringEntity(compact(render(json)), ContentType.APPLICATION_JSON)
+ val response = client.performRequest(
+ "POST",
+ s"/$index/$estype/_delete_by_query",
+ Map("refresh" -> ESUtils.getEventDataRefresh(config)).asJava)
+ val jsonResponse = parse(EntityUtils.toString(response.getEntity))
+ val result = (jsonResponse \ "result").extract[String]
+ result match {
+ case "deleted" => true
+ case _ =>
+ logger.error(s"[$result] Failed to update $index/$estype:$eventId")
false
}
+ } catch {
+ case e: IOException =>
+ logger.error(s"Failed to update $index/$estype:$eventId", e)
+ false
--- End diff --
Same as above.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio issue #421: Elasticsearch 5.x singleton client with a...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on the issue:
https://github.com/apache/incubator-predictionio/pull/421
> 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...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio issue #421: Elasticsearch 5.x singleton client with a...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on the issue:
https://github.com/apache/incubator-predictionio/pull/421
@mars LGTM! but sorry for conflict with my previous commit: 6789dbeb71b1cc7f13a385032da6fbc3b8cf7a12
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio issue #421: Elasticsearch 5.x singleton client with a...
Posted by mars <gi...@git.apache.org>.
Github user mars commented on the issue:
https://github.com/apache/incubator-predictionio/pull/421
I will resolve these conflicts today and then merge this PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134399019
--- Diff: core/src/main/scala/org/apache/predictionio/workflow/CleanupFunctions.scala ---
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.predictionio.workflow
+
+/** :: DeveloperApi ::
+ * Singleton object that collects anonymous functions to be
+ * executed to allow the process to end gracefully.
+ *
+ * For example, the Elasticsearch REST storage client
+ * maintains an internal connection pool that must
+ * be closed to allow the process to exit.
+ */
+object CleanupFunctions {
+ var functions: Seq[() => Unit] = Seq.empty[() => Unit]
+
+ /** Add a function to be called during cleanup.
+ *
+ * {{{
+ * import org.apache.predictionio.workflow.CleanupFunctions
+ *
+ * CleanupFunctions.add { MyStorageClass.close }
+ * }}}
+ *
+ * @param anonymous function containing cleanup code.
+ */
+ def add(f: () => Unit): Seq[() => Unit] = {
+ functions = functions :+ f
+ functions
+ }
+
+ /** Call all cleanup functions in order added.
+ *
+ * {{{
+ * import org.apache.predictionio.workflow.CleanupFunctions
+ *
+ * try {
+ * // Much code that needs cleanup
+ * // whether successful or error thrown.
+ * } finally {
+ * CleanupFunctions.run()
+ * }
+ * }}}
+ *
+ * @param anonymous function containing cleanup code.
+ */
+ def run(): Unit = {
+ functions.map { case f => f() }
--- End diff --
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:
```scala
functions.foreach { f => f() }
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134419406
--- Diff: storage/elasticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/ESPEvents.scala ---
@@ -107,37 +107,32 @@ class ESPEvents(client: ESClient, config: StorageClientConfig, index: String)
eventIds: RDD[String],
appId: Int, channelId: Option[Int])(sc: SparkContext): Unit = {
val estype = getEsType(appId, channelId)
- val restClient = client.open()
- try {
- eventIds.foreachPartition { iter =>
- iter.foreach { eventId =>
- try {
- val json =
- ("query" ->
- ("term" ->
- ("eventId" -> eventId)))
- val entity = new NStringEntity(compact(render(json)), ContentType.APPLICATION_JSON)
- val response = restClient.performRequest(
- "POST",
- s"/$index/$estype/_delete_by_query",
- Map("refresh" -> ESUtils.getEventDataRefresh(config)).asJava)
- val jsonResponse = parse(EntityUtils.toString(response.getEntity))
- val result = (jsonResponse \ "result").extract[String]
- result match {
- case "deleted" => true
- case _ =>
- logger.error(s"[$result] Failed to update $index/$estype:$eventId")
- false
- }
- } catch {
- case e: IOException =>
- logger.error(s"Failed to update $index/$estype:$eventId", e)
+ eventIds.foreachPartition { iter =>
+ iter.foreach { eventId =>
+ try {
+ val json =
+ ("query" ->
+ ("term" ->
+ ("eventId" -> eventId)))
+ val entity = new NStringEntity(compact(render(json)), ContentType.APPLICATION_JSON)
+ val response = client.performRequest(
+ "POST",
+ s"/$index/$estype/_delete_by_query",
+ Map("refresh" -> ESUtils.getEventDataRefresh(config)).asJava)
+ val jsonResponse = parse(EntityUtils.toString(response.getEntity))
+ val result = (jsonResponse \ "result").extract[String]
+ result match {
+ case "deleted" => true
+ case _ =>
+ logger.error(s"[$result] Failed to update $index/$estype:$eventId")
false
--- End diff --
Same as above.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio issue #421: Elasticsearch 5.x singleton client with a...
Posted by mars <gi...@git.apache.org>.
Github user mars commented on the issue:
https://github.com/apache/incubator-predictionio/pull/421
Based on these [Scala Concurrency/Thread Safety docs](https://twitter.github.io/scala_school/concurrency.html#danger), I believe simply annotating `@volatile` will cause the synchronization needed for thread-safety in this case.
So, I updated this PR with that change.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134419379
--- Diff: storage/elasticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/ESPEvents.scala ---
@@ -107,37 +107,32 @@ class ESPEvents(client: ESClient, config: StorageClientConfig, index: String)
eventIds: RDD[String],
appId: Int, channelId: Option[Int])(sc: SparkContext): Unit = {
val estype = getEsType(appId, channelId)
- val restClient = client.open()
- try {
- eventIds.foreachPartition { iter =>
- iter.foreach { eventId =>
- try {
- val json =
- ("query" ->
- ("term" ->
- ("eventId" -> eventId)))
- val entity = new NStringEntity(compact(render(json)), ContentType.APPLICATION_JSON)
- val response = restClient.performRequest(
- "POST",
- s"/$index/$estype/_delete_by_query",
- Map("refresh" -> ESUtils.getEventDataRefresh(config)).asJava)
- val jsonResponse = parse(EntityUtils.toString(response.getEntity))
- val result = (jsonResponse \ "result").extract[String]
- result match {
- case "deleted" => true
- case _ =>
- logger.error(s"[$result] Failed to update $index/$estype:$eventId")
- false
- }
- } catch {
- case e: IOException =>
- logger.error(s"Failed to update $index/$estype:$eventId", e)
+ eventIds.foreachPartition { iter =>
+ iter.foreach { eventId =>
+ try {
+ val json =
+ ("query" ->
+ ("term" ->
+ ("eventId" -> eventId)))
+ val entity = new NStringEntity(compact(render(json)), ContentType.APPLICATION_JSON)
+ val response = client.performRequest(
+ "POST",
+ s"/$index/$estype/_delete_by_query",
+ Map("refresh" -> ESUtils.getEventDataRefresh(config)).asJava)
+ val jsonResponse = parse(EntityUtils.toString(response.getEntity))
+ val result = (jsonResponse \ "result").extract[String]
+ result match {
+ case "deleted" => true
--- End diff --
`true` at here is no effect because the return type us `Unit`. I know it came from original code, but we should remove it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134416803
--- Diff: storage/elasticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/StorageClient.scala ---
@@ -18,27 +18,84 @@
package org.apache.predictionio.data.storage.elasticsearch
import org.apache.http.HttpHost
+import org.apache.http.auth.{AuthScope, UsernamePasswordCredentials}
+import org.apache.http.impl.client.BasicCredentialsProvider
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder
import org.apache.predictionio.data.storage.BaseStorageClient
import org.apache.predictionio.data.storage.StorageClientConfig
import org.apache.predictionio.data.storage.StorageClientException
+import org.apache.predictionio.workflow.CleanupFunctions
import org.elasticsearch.client.RestClient
+import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback
import grizzled.slf4j.Logging
-case class ESClient(hosts: Seq[HttpHost]) {
- def open(): RestClient = {
+object ESClient extends Logging {
+ var _sharedRestClient: Option[RestClient] = None
--- End diff --
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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by mars <gi...@git.apache.org>.
Github user mars commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134630517
--- Diff: storage/elasticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/StorageClient.scala ---
@@ -18,27 +18,84 @@
package org.apache.predictionio.data.storage.elasticsearch
import org.apache.http.HttpHost
+import org.apache.http.auth.{AuthScope, UsernamePasswordCredentials}
+import org.apache.http.impl.client.BasicCredentialsProvider
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder
import org.apache.predictionio.data.storage.BaseStorageClient
import org.apache.predictionio.data.storage.StorageClientConfig
import org.apache.predictionio.data.storage.StorageClientException
+import org.apache.predictionio.workflow.CleanupFunctions
import org.elasticsearch.client.RestClient
+import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback
import grizzled.slf4j.Logging
-case class ESClient(hosts: Seq[HttpHost]) {
- def open(): RestClient = {
+object ESClient extends Logging {
+ var _sharedRestClient: Option[RestClient] = None
--- End diff --
Thanks for the hints here. I suspected this would be an issue. I'm investigating how to make this threadsafe.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134417513
--- Diff: storage/elasticsearch/src/main/scala/org/apache/predictionio/data/storage/elasticsearch/StorageClient.scala ---
@@ -18,27 +18,84 @@
package org.apache.predictionio.data.storage.elasticsearch
import org.apache.http.HttpHost
+import org.apache.http.auth.{AuthScope, UsernamePasswordCredentials}
+import org.apache.http.impl.client.BasicCredentialsProvider
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder
import org.apache.predictionio.data.storage.BaseStorageClient
import org.apache.predictionio.data.storage.StorageClientConfig
import org.apache.predictionio.data.storage.StorageClientException
+import org.apache.predictionio.workflow.CleanupFunctions
import org.elasticsearch.client.RestClient
+import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback
import grizzled.slf4j.Logging
-case class ESClient(hosts: Seq[HttpHost]) {
- def open(): RestClient = {
+object ESClient extends Logging {
+ var _sharedRestClient: Option[RestClient] = None
+
+ def open(
+ hosts: Seq[HttpHost],
+ basicAuth: Option[(String, String)] = None): RestClient = {
try {
- RestClient.builder(hosts: _*).build()
+ val newClient = _sharedRestClient match {
+ case Some(c) => c
+ case None => {
+ var builder = RestClient.builder(hosts: _*)
+ builder = basicAuth match {
+ case Some((username, password)) => builder.setHttpClientConfigCallback(
+ new BasicAuthProvider(username, password))
+ case None => builder}
+ builder.build()
+ }
+ }
+ _sharedRestClient = Some(newClient)
+ newClient
} catch {
case e: Throwable =>
throw new StorageClientException(e.getMessage, e)
}
}
+
+ def close(): Unit = {
+ if (!_sharedRestClient.isEmpty) {
+ _sharedRestClient.get.close()
+ _sharedRestClient = None
+ }
--- End diff --
Typically, `Option.get()` is hated in the Scala world. You can write as follows instead:
```scala
_sharedRestClient.foreach { client =>
client.close()
_sharedRestClient = None
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134398339
--- Diff: core/src/main/scala/org/apache/predictionio/workflow/CleanupFunctions.scala ---
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.predictionio.workflow
+
+/** :: DeveloperApi ::
+ * Singleton object that collects anonymous functions to be
+ * executed to allow the process to end gracefully.
+ *
+ * For example, the Elasticsearch REST storage client
+ * maintains an internal connection pool that must
+ * be closed to allow the process to exit.
+ */
+object CleanupFunctions {
+ var functions: Seq[() => Unit] = Seq.empty[() => Unit]
--- End diff --
This field should be private or protected at least.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio issue #421: Elasticsearch 5.x singleton client with a...
Posted by mars <gi...@git.apache.org>.
Github user mars commented on the issue:
https://github.com/apache/incubator-predictionio/pull/421
Cheers @takezoe I addressed all your Scala style & usage suggestions. Still need to take care of the threadsafety issue with the singleton client.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by takezoe <gi...@git.apache.org>.
Github user takezoe commented on a diff in the pull request:
https://github.com/apache/incubator-predictionio/pull/421#discussion_r134399760
--- Diff: 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 = {
--- End diff --
You can get stacktrace from an exception using `ExceptionUtils` of commons-lang3.
```scala
import org.apache.commons.lang3.exception.ExceptionUtils
val stackTrace = ExceptionUtils.getStackTrace(e)
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-predictionio pull request #421: Elasticsearch 5.x singleton client...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-predictionio/pull/421
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---