Closes: #190 #189

Working on some tweaks to the request pipeline to make it nicer to use. I'll experiment in here for a bit and try get the abstractions nicer.

I'm receiving the following error:

ERROR es-install :: Client(ClientError { inner: Error(Request, State { next_error: Some(Error(Hyper(Error(Connect, Custom { kind: InvalidInput, error: NotHttp })), "https://localhost:9201/mt-ledger?pretty=true")), backtrace: None }) })

Here is the code which produces this error:

pub mod client {
    use elastic::{
        client::{prelude::SyncClient, SyncClientBuilder},
        error::Error,
    };
    use std::result::Result;
    pub fn make() -> Result<SyncClient, Error> {
        SyncClientBuilder::new()
            .static_node("https://localhost:9201")
            .params_fluent(move |p| p.url_param("pretty", true))
            .build()
    }
}

use mtledger::elasticsearch::client as EsClient;

fn main() {
    let client = EsClient::make().expect("Failed to create ES client");

    match client.ping().send() {
        Ok(response) => {
            println!("PING :: {:?}", response);
        }
        Err(error) => {
            println!("ERROR es-ping :: {:?}", error);
        }
    }
}

How do I connect to an HTTPS node?

Hi, I've used your crate successfully to query ES - it works great, do you have any supporting crates (or plans for) to assist with processing the output/returns?

I've spent quite a bit of time over the last few years parsing, iterating over results (aggregations as well as normal search hits), and am about to start playing with implementing an actual rusty Iterator for the results returned by your library. The idea is that it would facilitate hassle-free, fast (zero-copy) processing that can use rust's built in goodness (https://doc.rust-lang.org/beta/std/iter/index.html), as well as something like Rayon (https://github.com/nikomatsakis/rayon).

Would love to get your thought on this, and if its something that you would be interested in as a contribution, where it should be plugged into this eco-system.

Cheers,

Long term, it looks like the new rotor_http might be a good base for building the HTTP off.

https://github.com/tailhook/rotor-http/blob/master/src/client/request.rs

Tasks

• [x] Http GET connection to /_search
• [x] Http POST connection to /_search with body
• [x] Loop in separate thread that takes commands via Notifier
• [x] Persistent connection to /_search with various bodies

This is a bit of a "wild" idea, but something worth considering for the future. In a larger system with lots of users it is really convenient to externalize queries into files, although the facility exists to define queries through structured Rust, there is something to be said for being able to prototype in Kibana or Sense and then take copy the query for use in other projects, this proposal would see us have:

1. convenient method to load ES queries from files (say myquery.es), this is probably 3 lines of code of a good day, but might still make sense to have available to users.
2. template pre-processing of query strings, use a crate such as tera to process raw queries just before submission which allows: updating a timestamp, swapping out a term field, etc, etc. This works surprisingly well for use-cases where it makes more sense to have queries stored as strings (perhaps in files)

I would like to propose a breaking change that would allow the library to work with Elasticsearch 7 where the document type was removed after being deprecated in version 6.

The change consists mostly of defaulting to the _doc type instead of doc. By doing this, most ES routes work as intended.

I'm using this in one of my applications that uses the generated mapping and indexes documents just fine.

Note that while it would be possible to manually specify the _doc type everywhere, the mapping API should still be changed from PUT /my-index/_mappings/_doc to PUT /my-index/_mapping.

By adding Elastic dependency, whole cargo process breaks

~/.cargo/bin/cargo build
    Updating registry `https://github.com/rust-lang/crates.io-index`

Runs forever. Removing this dependency, everything starts working fine

Hello,

When running the index.rs example, I receive the following error:

thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Error(Response(Json(ErrorImpl { code: ExpectedSomeValue, line: 1, column: 1 })), State { next_error: None, backtrace: None })', /checkout/src/libcore/result.rs:860:4
note: Run with `RUST_BACKTRACE=1` for a backtrace.

The only change I have made to the example is adding the base URL for my own cluster using the base_url() function.

I am using version 0.13.1 of elastic.

I am happy to provide the full backtrace if it is helpful.

Thanks for your time!

TODO

• [x] Update CI to use new paths and run integration tests (on nightly? Or separate job?)
• [x] Update all examples
• [x] Fix current doc tests
• [x] Add new doc tests for new items
• [x] Update all links in docs

Closes:

• #195
• #212
• #223
• #214
• #226
• #167

Hopefully there won't be many more giant PRs like this for a while as the API starts to settle.

Motivation

I've rolled a few key API changes into this PR:

• Support a futures-based asynchronous client without having to fork too much of the request code.
• Provide a nicer story for error handling

Design

Async

The current idea is to use a generic type to distinguish a synchronous from an asynchronous client. This looks something like:

trait Sender {
    type Http;
    type Body;
}

impl Sender for SyncSender {
    type Http = reqwest::Client;
    type Body = elastic_reqwest::SyncBody;
}

impl Sender for AsyncSender {
    type Http = reqwest::unstable::async::Client;
    type Body = elastic_reqwest::AsyncBody;
}

struct Client<TSender: Sender> {
    http: TSender::Http
}

type SyncClient = Client<SyncSender>;
type AsyncClient = Client<AsyncSender>;

This means request builders can implement different behaviour depending on the kind of sender, or implement behaviour that's shared by all senders:

struct RequestBuilder<TSender, TRequest> {
    client: Client<TSender>,
    inner: TRequest
}

// Behaviour for either a sync or async Sender
impl<TSender: Sender> RequestBuilder<TSender, SearchRequestBuilder> {
    fn index(self, index: Index) -> Self {
        ...
    }
}

// Behaviour for a sync Sender
impl RequestBuilder<SyncSender, SearchRequestBuilder> {
    fn send(self) -> Result<SearchResponse> {
        ...
    }
}

// Behaviour for an async Sender
impl RequestBuilder<AsyncSender, SearchRequestBuilder> {
    fn send(self) -> impl Future<Item = SearchResponse> {
        ...
    }
}

The responses are currently not generic over the Sender, we've just got a SyncResponseBuilder and AsyncResponseBuilder. This was easier than trying to make them generic but is a bit of an asymmetrical wart on the API.

Benefits

• We don't have to duplicate request code, and it stays almost the same as it already is. We just pepper some generics around and add async implementations

Drawbacks

• It's quite complex and makes the rustdoc harder to follow.
• The Sender type is a bit of a false abstraction, since it's not really useful apart from choosing one of two implementations. You can't reasonably go and implement Sender yourself and have everything work. It'll need some kind of Sealed supertrait if it was going to stay the same. Otherwise we'll need to pull all necessary types and callbacks into the Sender and thread that through the whole app, rather than forking sync and async when sending.
• Responses are handled totally differently to requests. I think they should be roughly the same.

Error Handling

Offer errors types that:

• can be extended without requiring a breaking change
• are easily matchable when the error is likely recoverable
• provide detailed output when the error is likely non-recoverable

Right now, I'm making the assumption that:

• API errors are likely recoverable
• Any other kind of error (connection refused, invalid JSON, too many redirects etc) are likely non-recoverable.

The new error type looks roughly like:

pub enum Error {
    Api(ApiError),
    Client(ClientError),
    #[doc(hidden)]
    __NonExhaustive
}

pub struct ClientError {
    inner: inner::Error
}

mod inner {
    error_chain! {
        Build,
        Request,
        Response(u16),
    }
}

The idea is:

• Any error can be converted into any ClientError from within the elastic crate. The inner error is simply boxed along with a backtrace by error-chain.
• API errors are plain structures that can be matched and destructured
• Client errors are black boxes that can be logged, but not matched

Benefits

• We've got freedom to tweak error types in elastic_reqwest, elastic_responses etc without affecting the error handling upstream
• We don't have lots of duplicate error variants to manage

Drawbacks

• You can't do much with client errors besides log them. Maybe there are cases a client would like to know about individually and cater for
• Each client error means boxing
• We might be holding error-chain in a way it wasn't meant to be by hiding it away. Backtraces might get repeated and be difficult to follow

Integration tests

Add some infrastructure for spinning up an Elasticsearch node in a docker container and running a suite of idempotent integration tests against it. This should support different Elasticsearch/client configurations, such as requiring authentication, TLS, etc.

This project is mostly about talking to Elasticsearch, so we need good confidence that it actually works.

Having some ad-hoc integration tests is enough to get started, but using the YAML spec at some point in the future would also be good.

This might be really easy but I can't figure it out.

The following:

struct ID(String);

impl KeywordFieldType<DefaultKeywordMapping> for ID {}

maps to:

"id": {
    "type": "text",
    "fields": {
        "keyword": {
            "type": "keyword",
            "ignore_above": 256
        }
    }
}

I want it to map to:

"id": {
    "type": "keyword"
}

so that you don't have to specify id.keyword when searching...

What am I missing?

For good measure I have also tried:

struct IdMapping;

impl KeywordMapping for IdMapping {
    fn analyzer() -> Option<&'static str> {
        Some("keyword")
    }
}

impl KeywordFieldType<IdMapping> for ID {}

This might be related to #330...

version: 0.21.0-pre.1

If I index documents using bulk (and probably with index etc. too) like:

#[derive(Serialize, Deserialize, ElasticType, Debug)]
#[elastic(index = "test")]
pub struct MyType {
    timestamp: ChronoDateTime,
}

let ops = (0..1000).into_iter().map(|i| {
    bulk::<MyType>().index(MyType {
        timestamp: "2019-02-18T00:00:00Z".parse().unwrap(),
    })
});

let response = client.bulk().extend(ops).send()?;

querying the new documents via search or get like:

client
    .search::<MyType>()
    .body(json!({ "query": { "match_all": {} } }))
    .send()?;

produces an error:

error sending a request or receiving a response. 
Caused by: error receiving a response. 
Status code: 200 OK

However, manually building the request works:

client
    .request(SimpleSearchRequest::for_index_ty(MyType::static_index(), MyType::static_ty()))
    .send()?
    .into_response::<SearchResponse<Value>>()?;

Also, not sure if this is supposed to work, but it produces the same error:

client
    .request(SimpleSearchRequest::for_index_ty(MyType::static_index(), MyType::static_ty()))
    .send()?
    .into_response::<SearchResponse<MyType>>()?;

My code starting on line 169:

let es = elastic::client::AsyncClientBuilder::new()
    .static_nodes(backend_url)
    .build()
    .unwrap(); // TODO
let timestamp = Utc::now();
let stats = Stats {
    core: "helsinki.fi".to_string(),
    status: res.response().status().as_u16(),
    query_time: entry_time.elapsed().whole_milliseconds(),
    timestamp: timestamp
        .format("%Y-%m-%d %H:%M:%S%.6f%:z")
        .to_string(),
    // TODO: remove stopwords
    original_query: clean_and_split_query(&q.1),
    query: clean_and_split_query(&q.1),
    stopword_count: 0,
    removed_stopwords: vec![],
};
let index_name = format!("searches-{}", timestamp.format("%Y%m%d"));
let send_fut = es.document().index_raw(index_name, stats).send();
send_fut.then(|_| ());

My error message:

   --> src/logging.rs:189:34
    |
189 |                         send_fut.then(|_| Ok(()));
    |                                  ^^^^ method not found in `elastic::client::requests::document_index::Pending`
    |
    = help: items from traits can only be used if the trait is in scope
    = note: the following trait is implemented but not in scope; perhaps add a `use` for it:
            `use futures::future::Future;`

I do have this use line in my file:

use futures::future::{ok, Ready, Future};

The default mapping type for serde_json::Value is "nested" and that not allow make normal queries on ElasticSearch for dynamics data. How i can change the type to "object". Already i changed the code on file "src/elastic/src/types/document/mapping.rs" on trait ObjectMapping from NESTED_DATATYPE to OBJECT_DATATYPE and now my elastic queries are responding with results.

This seems closely related to https://github.com/elastic-rs/elastic/issues/377 but opening a new issues as that one is closed.

Getting "expected struct HitsTotal" issue with this on elasticsearch 6.7 (and the following versions for both elastic and elastic_derive crates).

• 0.21.0-pre.2 ☑️ Good
• 0.21.0-pre.3 ☑️ Good
• 0.21.0-pre.4 ☑️ Good
• 0.21.0-pre.5 ❌ Regression

(Parse(ParseError { inner: Error(\"invalid type: integer `0`, expected struct HitsTotal\", line: 11, column: 16) }))

#[derive(Clone, Debug, Serialize, Deserialize, ElasticType)]
pub struct PortfolioIdListing {
    pub id: String,
    pub idlisting: types::IdListing,
}

let response = esc
    .search::<es_models::PortfolioIdListing>()
    .index(EsIndex::Portfolio.name())
    .body(json!({
        "query": {
            "bool": {
                "must": {
                        // -1 is what the JS import script sets them to by default
                    "term": { "pf_likes": -1 },
                },
                "must_not": {
                    "terms": { "idlisting": exclude },
                },
            }
        },
        "size": Value::Number(BATCH_SIZE.into()),
        "_source": ["id", "idlisting"]
    }))
    .send()

Hey, thanks for making this!

In the docs it's noted that you can specify index expressions, however, turning this on results in a huge pile of the trait `elastic::prelude::StaticIndex` is not implemented for ... errors as afaik many of the library methods are implemented over StaticIndex, and it's not super clear how to fix this?

An example is put_mapping, which requires StaticIndex to create a PutMappingRequestBuilder, unfortunately the inner type is private so unless the type fulfills StaticIndex it is,, impossible (?), to set a mapping?

Whenever I try to make a bulk query with the following code my program crashes with a stack overflow exception:

let ops = all_bets.into_iter().map(|x| bulk::<BetSummary>().index(x));                         
client.bulk().index(BetSummary::index()).ty(BetSummary::static_ty()).extend(ops).send()?;  

Could you let me know what I am doing wrong?