Skip to content

Commit da0dbe2

Browse files
To-omTo-om
committed
#82 Fix ElasticSearch configuraiton
1 parent 16c337b commit da0dbe2

File tree

4 files changed

+84
-54
lines changed

4 files changed

+84
-54
lines changed

‎app/org/elastic4play/database/DBConfiguration.scala

+55-39
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,10 @@ import com.sksamuel.elastic4s.streams.ReactiveElastic.ReactiveElastic
2222
import com.sksamuel.elastic4s.streams.{RequestBuilder, ResponseListener}
2323
import javax.inject.{Inject, Named, Singleton}
2424
import javax.net.ssl.{KeyManagerFactory, SSLContext, TrustManagerFactory}
25+
import org.apache.http.auth.{AuthScope, UsernamePasswordCredentials}
26+
import org.apache.http.client.CredentialsProvider
2527
import org.apache.http.client.config.RequestConfig
28+
import org.apache.http.impl.client.BasicCredentialsProvider
2629
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder
2730
import org.elasticsearch.client.RestClientBuilder.{HttpClientConfigCallback, RequestConfigCallback}
2831

@@ -43,8 +46,8 @@ class DBConfiguration @Inject()(
4346
) {
4447
private[DBConfiguration] lazy val logger = Logger(getClass)
4548

46-
def requestConfig: RequestConfigCallback = (requestConfigBuilder: RequestConfig.Builder) ⇒ {
47-
config.getOptional[Boolean]("search.authenticationEnabled").foreach(requestConfigBuilder.setAuthenticationEnabled)
49+
def requestConfigCallback: RequestConfigCallback = (requestConfigBuilder: RequestConfig.Builder) ⇒ {
50+
requestConfigBuilder.setAuthenticationEnabled(credentialsProviderMaybe.isDefined)
4851
config.getOptional[Boolean]("search.circularRedirectsAllowed").foreach(requestConfigBuilder.setCircularRedirectsAllowed)
4952
config.getOptional[Int]("search.connectionRequestTimeout").foreach(requestConfigBuilder.setConnectionRequestTimeout)
5053
config.getOptional[Int]("search.connectTimeout").foreach(requestConfigBuilder.setConnectTimeout)
@@ -62,53 +65,66 @@ class DBConfiguration @Inject()(
6265
requestConfigBuilder
6366
}
6467

65-
def httpClientConfig: HttpClientConfigCallback = (httpClientBuilder: HttpAsyncClientBuilder) ⇒ {
66-
config.getOptional[String]("search.keyStore.path").map { keyStore ⇒
67-
val keyStorePath = Paths.get(keyStore)
68-
val keyStoreType = config.getOptional[String]("search.keyStore.type").getOrElse(KeyStore.getDefaultType)
69-
val keyStorePassword = config.getOptional[String]("search.keyStore.password").getOrElse("").toCharArray
70-
val keyInputStream = Files.newInputStream(keyStorePath)
71-
val keyManagers = try {
72-
val keyStore = KeyStore.getInstance(keyStoreType)
73-
keyStore.load(keyInputStream, keyStorePassword)
74-
val kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm)
75-
kmf.init(keyStore, keyStorePassword)
76-
kmf.getKeyManagers
77-
} finally {
78-
keyInputStream.close()
79-
}
68+
lazy val credentialsProviderMaybe: Option[CredentialsProvider] =
69+
for {
70+
user ← config.getOptional[String]("search.user")
71+
password ← config.getOptional[String]("search.password")
72+
} yield {
73+
val provider = new BasicCredentialsProvider
74+
val credentials = new UsernamePasswordCredentials(user, password)
75+
provider.setCredentials(AuthScope.ANY, credentials)
76+
provider
77+
}
78+
79+
lazy val sslContextMaybe: Option[SSLContext] = config.getOptional[String]("search.keyStore.path").map { keyStore ⇒
80+
val keyStorePath = Paths.get(keyStore)
81+
val keyStoreType = config.getOptional[String]("search.keyStore.type").getOrElse(KeyStore.getDefaultType)
82+
val keyStorePassword = config.getOptional[String]("search.keyStore.password").getOrElse("").toCharArray
83+
val keyInputStream = Files.newInputStream(keyStorePath)
84+
val keyManagers = try {
85+
val keyStore = KeyStore.getInstance(keyStoreType)
86+
keyStore.load(keyInputStream, keyStorePassword)
87+
val kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm)
88+
kmf.init(keyStore, keyStorePassword)
89+
kmf.getKeyManagers
90+
} finally {
91+
keyInputStream.close()
92+
}
8093

81-
val trustManagers = config
82-
.getOptional[String]("search.trustStore.path")
83-
.map { trustStorePath ⇒
84-
val keyStoreType = config.getOptional[String]("search.keyStore.type").getOrElse(KeyStore.getDefaultType)
85-
val trustStorePassword = config.getOptional[String]("search.trustStore.password").getOrElse("").toCharArray
86-
val trustInputStream = Files.newInputStream(Paths.get(trustStorePath))
87-
try {
88-
val keyStore = KeyStore.getInstance(keyStoreType)
89-
keyStore.load(trustInputStream, trustStorePassword)
90-
val tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm)
91-
tmf.init(keyStore)
92-
tmf.getTrustManagers
93-
} finally {
94-
trustInputStream.close()
95-
}
94+
val trustManagers = config
95+
.getOptional[String]("search.trustStore.path")
96+
.map { trustStorePath ⇒
97+
val keyStoreType = config.getOptional[String]("search.trustStore.type").getOrElse(KeyStore.getDefaultType)
98+
val trustStorePassword = config.getOptional[String]("search.trustStore.password").getOrElse("").toCharArray
99+
val trustInputStream = Files.newInputStream(Paths.get(trustStorePath))
100+
try {
101+
val keyStore = KeyStore.getInstance(keyStoreType)
102+
keyStore.load(trustInputStream, trustStorePassword)
103+
val tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm)
104+
tmf.init(keyStore)
105+
tmf.getTrustManagers
106+
} finally {
107+
trustInputStream.close()
96108
}
97-
.getOrElse(Array.empty)
109+
}
110+
.getOrElse(Array.empty)
98111

99-
// Configure the SSL context to use TLS
100-
val sslContext = SSLContext.getInstance("TLS")
101-
sslContext.init(keyManagers, trustManagers, null)
112+
// Configure the SSL context to use TLS
113+
val sslContext = SSLContext.getInstance("TLS")
114+
sslContext.init(keyManagers, trustManagers, null)
115+
sslContext
116+
}
102117

103-
httpClientBuilder.setSSLContext(sslContext)
104-
}
118+
def httpClientConfig: HttpClientConfigCallback = (httpClientBuilder: HttpAsyncClientBuilder) ⇒ {
119+
sslContextMaybe.foreach(httpClientBuilder.setSSLContext)
120+
credentialsProviderMaybe.foreach(httpClientBuilder.setDefaultCredentialsProvider)
105121
httpClientBuilder
106122
}
107123

108124
/**
109125
* Underlying ElasticSearch client
110126
*/
111-
private[database] val client = ElasticClient(ElasticProperties(config.get[String]("search.uri")), requestConfig, httpClientConfig)
127+
private[database] val client = ElasticClient(ElasticProperties(config.get[String]("search.uri")), requestConfigCallback, httpClientConfig)
112128
// when application close, close also ElasticSearch connection
113129
lifecycle.addStopHook { () ⇒
114130
Future {

‎app/org/elastic4play/database/DBCreate.scala

-1
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,6 @@ import com.sksamuel.elastic4s.indexes.IndexRequest
1313
import com.sksamuel.elastic4s.streams.RequestBuilder
1414
import javax.inject.{Inject, Singleton}
1515

16-
import org.elastic4play.CreateError
1716
import org.elastic4play.models.BaseEntity
1817

1918
/**

‎app/org/elastic4play/database/DBFind.scala

+3-3
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,7 @@ class DBFind(pageSize: Int, keepAlive: FiniteDuration, db: DBConfiguration, impl
114114
db.execute(searchRequest)
115115
.recoverWith {
116116
case t if t == IndexNotFoundException ⇒ Future.failed(t)
117-
case t ⇒ Future.failed(SearchError("Invalid search query", t))
117+
case t ⇒ Future.failed(SearchError("Invalid search query"))
118118
}
119119
}
120120
}
@@ -173,7 +173,7 @@ class SearchWithScroll(db: DBConfiguration, SearchRequest: SearchRequest, keepAl
173173
val callback = getAsyncCallback[Try[SearchResponse]] {
174174
case Success(searchResponse) if searchResponse.isTimedOut ⇒
175175
logger.warn("Search timeout")
176-
failStage(SearchError("Request terminated early or timed out", null))
176+
failStage(SearchError("Request terminated early or timed out"))
177177
case Success(searchResponse) if searchResponse.isEmpty ⇒
178178
completeStage()
179179
case Success(searchResponse) if skip > 0 ⇒
@@ -190,7 +190,7 @@ class SearchWithScroll(db: DBConfiguration, SearchRequest: SearchRequest, keepAl
190190
pushNextHit()
191191
case Failure(error) ⇒
192192
logger.warn("Search error", error)
193-
failStage(SearchError("Request terminated early or timed out", error))
193+
failStage(SearchError("Request terminated early or timed out"))
194194
}
195195
val futureSearchResponse = scrollId.flatMap(s ⇒ db.execute(searchScroll(s).keepAlive(keepAliveStr)))
196196
scrollId = futureSearchResponse.map(_.scrollId.get)

‎app/org/elastic4play/services/MigrationSrv.scala

+26-11
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,27 @@
11
package org.elastic4play.services
22

3-
import javax.inject.{Inject, Singleton}
4-
53
import scala.concurrent.{ExecutionContext, Future}
64
import scala.util.{Failure, Success}
75

8-
import play.api.{Configuration, Logger}
96
import play.api.libs.json.JsValue.jsValueToJsLookup
107
import play.api.libs.json.Json.toJsFieldJsValueWrapper
118
import play.api.libs.json._
9+
import play.api.{Configuration, Logger}
1210

1311
import akka.NotUsed
1412
import akka.actor.ActorSystem
15-
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
1613
import akka.stream.scaladsl.{Sink, Source}
14+
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
1715
import com.sksamuel.elastic4s.http.ElasticDsl._
1816
import com.typesafe.config.Config
17+
import javax.inject.{Inject, Singleton}
1918

2019
import org.elastic4play.InternalError
2120
import org.elastic4play.database._
2221

2322
case class MigrationEvent(modelName: String, current: Long, total: Long) extends EventMessage
24-
case object EndOfMigrationEvent extends EventMessage
23+
24+
case object EndOfMigrationEvent extends EventMessage
2525

2626
object IndexType extends Enumeration {
2727
val indexWithMappingTypes, indexWithoutMappingTypes = Value
@@ -115,6 +115,7 @@ class MigrationSrv @Inject()(
115115
("_version" → JsNumber(hit.version))
116116
}
117117
}
118+
118119
override def count(tableName: String): Future[Long] =
119120
db.execute {
120121
search(db.indexName / tableName).matchAllQuery().size(0)
@@ -123,6 +124,7 @@ class MigrationSrv @Inject()(
123124
_.totalHits
124125
}
125126
.recover { case _ ⇒ 0L }
127+
126128
override def getEntity(tableName: String, entityId: String): Future[JsObject] = currentdbget(tableName, entityId)
127129
}
128130

@@ -170,7 +172,10 @@ class MigrationSrv @Inject()(
170172
val r = entities
171173
.zipWith(count) { (entity, current) ⇒
172174
eventSrv.publish(MigrationEvent(modelName, current.toLong, total))
173-
entity
175+
(entity \ "_type").asOpt[JsString].fold(entity) { t ⇒
176+
val relations = (entity \ "_parent").asOpt[JsString].fold[JsValue](t)(p ⇒ Json.obj("name" → t, "parent" → p))
177+
entity - "_type" - "_parent" + ("relations" → relations)
178+
}
174179
}
175180
.runWith(dbcreate.sink())
176181
r.onComplete { x ⇒
@@ -230,6 +235,7 @@ class MigrationSrv @Inject()(
230235
def isMigrating: Boolean = !migrationProcess.isCompleted
231236
def isReady: Boolean = dbindex.indexStatus && migrationProcess.isCompleted
232237
}
238+
233239
/* Operation applied to the previous state of the database to get next version */
234240
trait Operation extends ((String ⇒ Source[JsObject, NotUsed]) ⇒ (String ⇒ Source[JsObject, NotUsed]))
235241

@@ -259,7 +265,8 @@ object Operation {
259265
})
260266

261267
def mapEntity(tables: String*)(transform: JsObject ⇒ JsObject): Operation = mapEntity(tables.contains, transform)
262-
def apply(table: String)(transform: JsObject ⇒ JsObject): Operation = mapEntity(_ == table, transform)
268+
269+
def apply(table: String)(transform: JsObject ⇒ JsObject): Operation = mapEntity(_ == table, transform)
263270

264271
def removeEntity(tableFilter: String ⇒ Boolean, filter: JsObject ⇒ Boolean): Operation =
265272
Operation((f: String ⇒ Source[JsObject, NotUsed]) ⇒ {
@@ -268,7 +275,8 @@ object Operation {
268275
})
269276

270277
def removeEntity(tables: String*)(filter: JsObject ⇒ Boolean): Operation = removeEntity(tables.contains, filter)
271-
def removeEntity(table: String)(filter: JsObject ⇒ Boolean): Operation = removeEntity(_ == table, filter)
278+
279+
def removeEntity(table: String)(filter: JsObject ⇒ Boolean): Operation = removeEntity(_ == table, filter)
272280

273281
def renameAttribute(tableFilter: String ⇒ Boolean, newName: String, oldNamePath: Seq[String]): Operation =
274282
Operation((f: String ⇒ Source[JsObject, NotUsed]) ⇒ {
@@ -283,6 +291,7 @@ object Operation {
283291

284292
def renameAttribute(tables: Seq[String], newName: String, oldNamePath: String*): Operation =
285293
renameAttribute(a ⇒ tables.contains(a), newName, oldNamePath)
294+
286295
def renameAttribute(table: String, newName: String, oldNamePath: String*): Operation = renameAttribute(_ == table, newName, oldNamePath)
287296

288297
def rename(value: JsObject, newName: String, path: Seq[String]): JsObject =
@@ -310,6 +319,7 @@ object Operation {
310319

311320
def mapAttribute(tables: Seq[String], attribute: String)(transform: JsValue ⇒ JsValue): Operation =
312321
mapAttribute(a ⇒ tables.contains(a), attribute, transform)
322+
313323
def mapAttribute(table: String, attribute: String)(transform: JsValue ⇒ JsValue): Operation = mapAttribute(_ == table, attribute, transform)
314324

315325
def removeAttribute(tableFilter: String ⇒ Boolean, attributes: String*): Operation =
@@ -320,8 +330,10 @@ object Operation {
320330
y - a
321331
}
322332
)
333+
323334
def removeAttribute(tables: Seq[String], attributes: String*): Operation = removeAttribute(a ⇒ tables.contains(a), attributes: _*)
324-
def removeAttribute(table: String, attributes: String*): Operation = removeAttribute(_ == table, attributes: _*)
335+
336+
def removeAttribute(table: String, attributes: String*): Operation = removeAttribute(_ == table, attributes: _*)
325337

326338
def addAttribute(tableFilter: String ⇒ Boolean, attributes: (String, JsValue)*): Operation =
327339
mapEntity(
@@ -331,8 +343,10 @@ object Operation {
331343
y + a
332344
}
333345
)
346+
334347
def addAttribute(tables: Seq[String], attributes: (String, JsValue)*): Operation = addAttribute(t ⇒ tables.contains(t), attributes: _*)
335-
def addAttribute(table: String, attributes: (String, JsValue)*): Operation = addAttribute(_ == table, attributes: _*)
348+
349+
def addAttribute(table: String, attributes: (String, JsValue)*): Operation = addAttribute(_ == table, attributes: _*)
336350

337351
def addAttributeIfAbsent(tableFilter: String ⇒ Boolean, attributes: (String, JsValue)*): Operation =
338352
mapEntity(tableFilter, { x ⇒
@@ -345,5 +359,6 @@ object Operation {
345359
})
346360

347361
def addAttributeIfAbsent(tables: Seq[String], attributes: (String, JsValue)*): Operation = addAttribute(t ⇒ tables.contains(t), attributes: _*)
348-
def addAttributeIfAbsent(table: String, attributes: (String, JsValue)*): Operation = addAttribute(_ == table, attributes: _*)
362+
363+
def addAttributeIfAbsent(table: String, attributes: (String, JsValue)*): Operation = addAttribute(_ == table, attributes: _*)
349364
}

0 commit comments

Comments
 (0)