-
Notifications
You must be signed in to change notification settings - Fork 112
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Keyword search] Index documents in elasticsearch (1/3) (#9384)
* new mapping * search store * index mapping fix * wip datasource * update create_index with store * impl clone in store * add the indexation call * add 200ms timeout * use Node::from * only index 1 on 5 * log indexation
- Loading branch information
1 parent
6bb3f7b
commit ec60d8a
Showing
6 changed files
with
130 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| use anyhow::Result; | ||
| use async_trait::async_trait; | ||
| use elasticsearch::{ | ||
| auth::Credentials, | ||
| http::transport::{SingleNodeConnectionPool, TransportBuilder}, | ||
| Elasticsearch, IndexParts, | ||
| }; | ||
| use rand::Rng; | ||
| use url::Url; | ||
|
|
||
| use crate::data_sources::node::Node; | ||
| use crate::{data_sources::data_source::Document, utils}; | ||
| use tracing::{error, info}; | ||
| #[async_trait] | ||
| pub trait SearchStore { | ||
| async fn index_document(&self, document: &Document) -> Result<()>; | ||
| fn clone_box(&self) -> Box<dyn SearchStore + Sync + Send>; | ||
| } | ||
|
|
||
| impl Clone for Box<dyn SearchStore + Sync + Send> { | ||
| fn clone(&self) -> Self { | ||
| self.clone_box() | ||
| } | ||
| } | ||
|
|
||
| #[derive(Clone)] | ||
| pub struct ElasticsearchSearchStore { | ||
| pub client: Elasticsearch, | ||
| } | ||
|
|
||
| impl ElasticsearchSearchStore { | ||
| pub async fn new(es_uri: &str, username: &str, password: &str) -> Result<Self> { | ||
| let credentials = Credentials::Basic(username.to_string(), password.to_string()); | ||
| let u = Url::parse(es_uri)?; | ||
| let conn_pool = SingleNodeConnectionPool::new(u); | ||
| let mut transport_builder = TransportBuilder::new(conn_pool); | ||
| transport_builder = transport_builder | ||
| .auth(credentials) | ||
| .disable_proxy() | ||
| .cert_validation(elasticsearch::cert::CertificateValidation::None); | ||
| let transport = transport_builder.build()?; | ||
| let client = Elasticsearch::new(transport); | ||
| Ok(Self { client }) | ||
| } | ||
| } | ||
|
|
||
| const NODES_INDEX_NAME: &str = "core.data_sources_nodes"; | ||
|
|
||
| #[async_trait] | ||
| impl SearchStore for ElasticsearchSearchStore { | ||
| async fn index_document(&self, document: &Document) -> Result<()> { | ||
| // elasticsearch needs to index a Node, not a Document | ||
| let node = Node::from(document.clone()); | ||
|
|
||
| // safety for rollout: we only index one time on five | ||
| // TODO(kw-search): remove this once prod testing is ok | ||
| if rand::thread_rng().gen_bool(0.8) { | ||
| return Ok(()); | ||
| } | ||
|
|
||
| // todo(kw-search): fail on error | ||
| let now = utils::now(); | ||
| match self | ||
| .client | ||
| .index(IndexParts::IndexId(NODES_INDEX_NAME, &document.document_id)) | ||
| .timeout("200ms") | ||
| .body(node) | ||
| .send() | ||
| .await | ||
| { | ||
| Ok(_) => { | ||
| info!( | ||
| duration = utils::now() - now, | ||
| document_id = document.document_id, | ||
| "[ElasticsearchSearchStore] Indexed document" | ||
| ); | ||
| Ok(()) | ||
| } | ||
| Err(e) => { | ||
| error!( | ||
| error = %e, | ||
| duration = utils::now() - now, | ||
| document_id = document.document_id, | ||
| "[ElasticsearchSearchStore] Failed to index document" | ||
| ); | ||
| Ok(()) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| fn clone_box(&self) -> Box<dyn SearchStore + Sync + Send> { | ||
| Box::new(self.clone()) | ||
| } | ||
| } |