OpenSearch with AWS Lambda Rust (part 1) - set up the local environment
Tags: aws, rust, elasticsearch
Published: 2024-03-01 20:22:50

#Introduction

Rust is well suited for creating low-latency lambda functions. It comes well with OpenSearch and lets build extremely performant integrations.

I plan to create a serverless application, that would use OpenSearch, and document my learnings along the way. I will create a few blog posts to keep things short and sweet.

#Project structure

The code for this blog post is [here](https://github.com/szymon-szym/opensearch-rust-lambda/tree/1-local-setup) (branch 1-local-setup)

##Dependencies

Infrastructure will be handled with AWS CDK.
For lambda-related tasks, I am going to use great [cargo lambda](https://www.cargo-lambda.info/)

##Architecture

To be honest I haven't decided yet how I want to use created lambdas. It is not crucial for the local setup though.

#Run OpenSearch locally and connect with the Rust client

##Folders structure

I would like to start development locally because there is a lot of stuff I can build and test before going to the cloud. I love to have a short feedback loop. 

Having that in mind I start by creating a root project as AWS CDK because eventually, I want to use it as my IaC.

```bash
cdk init --language typescript
```

Now I define a new folder `functions` and subfolders `query` and `opensearch_service`. `functions` is my root folder for lambda functions.

##Run OpenSearch locally

A local instance of OpenSearch is pretty handy. There are different ways to achieve it but for me, the easiest way to run OpenSearch locally  was to use `docker-compose.yml` from [official site](https://opensearch.org/docs/latest/install-and-configure/install-opensearch/docker/)

I just copied docker-compose file to the root folder of my project and ran `docker compose up`. I needed also to apply some [changes in the system configuration](https://opensearch.org/docs/latest/install-and-configure/install-opensearch/docker/#run-opensearch-in-a-docker-container), but it wasn't hard at all.

Once all containers are running, I can go to http://localhost:5601/app/home#/ and start playing with the OpenSearch dashboard (user: admin, password: admin)


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vn9u3hrv6neh61f7vjts.png)

That's nice!

I need data to test my functions, so I go to `Add sample data` and pick `Sample Flights Data` to be loaded


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/9ed841vqo0hjf1v9aymu.png)

Now I go to `dev tools` and confirm, that data was loaded


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/zntazwfpuz8b2dndq8jb.png)

My local OpenSearch instance is ready.

## Rust `OpenSearchService`

I plan to put functionalities related to OpenSearch in a separate library, so I can use them in different functions later on. 
I use the cargo feature which is a `workspace`. I can create a few crates in a single workspace (eg. different lambda functions, common libraries, etc.) 
In the `functions/` folder  I put `Cargo.toml` file with the following content:
```
[workspace]

members = [
    "query",
    "opensearch_service",
]
```


I go to the `functions/opensearch_service` folder and run `cargo init --lib`. Here I will place functionalities I want to be shared by different functions. For now, I need a client to connect to OpenSearch and perform a dummy query.

I will use the following dependencies: `opensearch, serde, serde_json, anyhow`. For me using `cargo add` is the most convenient way.

```bash
cargo add serde -F derive
cargo add opensearch -F aws-auth
cargo add anyhow serde_json
```

The `OpenSearchService` at this point is simple:

```rust
pub struct OpenSearchService {
    client: OpenSearch,    
}
```

I implement a function for creating a client with a local connection:

```rust
// ...
impl OpenSearchService {

    pub fn local_client() -> Self {
        let url = Url::parse("https://localhost:9200").unwrap();
        let conn_pool = SingleNodeConnectionPool::new(url);
        let credentials =
            opensearch::auth::Credentials::Basic("admin".to_string(), "admin".to_string());
        let cert_validation = opensearch::cert::CertificateValidation::None;
        let transport = TransportBuilder::new(conn_pool)
            .cert_validation(cert_validation)
            .auth(credentials)
            .build().unwrap();
        let client = OpenSearch::new(transport);

        Self { client }
    }
```

To be honest I struggled a lot with setting up this connection. It turned out that I needed to use a custom URL (the default one in use is http://) and configure the certificate validation flow. 

Once the client is ready, I build the function that runs an empty `must` query and parses the results:

```rust
// ...
// impl OpenSearchService {
// ... 
pub async fn query_all_docs<T>(&self, index: &str, limit: i64) -> anyhow::Result<Vec<T>> 
    where T: DeserializeOwned
    {
        let response = self.client
            .search(SearchParts::Index(&[index]))
            .size(limit)
            .from(0)
            .body(json!({
                "query": {
                    "bool": {
                        "must": []
                    }
                }
            }))
            .send()
            .await?;

        let response_body = response.json::<Value>().await?;

        let result = response_body["hits"]["hits"]
            .as_array()
            .unwrap()
            .iter()
            .map(|raw_value| {
                serde_json::from_value::<T>(raw_value["_source"].clone()).unwrap()
            })
            .collect::<Vec<_>>();

        Ok(result)
    }
```

The query function is generic, so it can be used for different types in a type-safe manner. A thing to remember is that we need to add some bounds for the generic type to make sure, that it can be deserialized `where T: DeserializeOwned`. If it is missing, the compiler in cooperation with stack overflow would help us close the gap.

Ok, the initial implementation of the `OpenSearchService` is in place, let's create a lambda function.

## Rust `query` lambda function

I go to the `functions/query` and run `cargo lambda init`. When prompted I select `no` for the question about creating http handler, and I leave the event type empty. 

In the `Cargo.toml` just created library can be added using the path
```
# ...
[dependencies]

opensearch_service = { path = "../opensearch_service"}
#...
```
I will use also `serde` and `serde_json`.

To be able to query my sample dataset I need to define a type for it. I create `FlightData` struct and copy fields from the OpenSearch dashboard.
In cases like this I usually just paste the commented json and start creating a struct - Code Whisperer is clever enough to figure out what I need


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/xlgqbwhw4t99i1mkykou.png)

Finally, the structure looks like this:

```rust
#[derive(Serialize, Deserialize, Debug)]
#[serde(rename_all = "PascalCase")]
pub struct FlightData {
    flight_num: String,
    dest_country: String,
    origin_weather: String,
    origin_city_name: String,
    avg_ticket_price: f32,
    distance_miles: f32,
    flight_delay: bool,
    dest_weather: String,
    dest: String,
    flight_delay_type: String,
    origin_country: String,
    #[serde(rename = "dayOfWeek")]
    day_of_week: u8,
    distance_kilometers: f32,
    #[serde(rename = "timestamp")]
    timestamp: String,
    dest_location: Location,
    #[serde(rename = "DestAirportID")]
    dest_airport_id: String,
    carrier: String,
    cancelled: bool,
    flight_time_min: f32,
    origin: String,
    origin_location: Location,
    dest_region: String,
    #[serde(rename = "OriginAirportID")]
    origin_airport_id: String,
    origin_region: String,
    dest_city_name: String,
    flight_time_hour: f32,
    flight_delay_min: i64,
}

#[derive(Serialize, Deserialize, Debug)]
pub struct Location {
    lat: String,
    lon: String,
}
```

For initial testing, request, and response can be simple:

```rust
#[derive(Deserialize)]
struct Request {
    limit: i64,
}

#[derive(Serialize)]
struct Response {
    flights: Vec<FlightData>,
}

async fn function_handler(event: LambdaEvent<Request>) -> Result<Response, Error> {
// ...
```

Once types are in place, lambda itself is pretty straightforward

```rust
// ...
async fn function_handler(os_client: &OpenSearchService, event: LambdaEvent<Request>) -> Result<Response, Error> {

    let limit = event.payload.limit;

    let index = "opensearch_dashboards_sample_data_flights";

    let result = os_client.query_all_docs::<FlightData>(index, limit).await?;
    // Prepare the response
    let resp = Response {
        flights: result,
    };

    Ok(resp)
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing::init_default_subscriber();

    let os_client = opensearch_service::OpenSearchService::local_client();


    run(service_fn(|event: LambdaEvent<Request>| {
        function_handler(&os_client, event)
    })).await
}
//...
```

## Testing lambda locally

Once all is prepared I run the function and see if it works. I create a dummy test event in the `functions/query/events/flights.json` file

```json
{
    "limit": 5
}
```

Testing lambda locally with cargo lambda is a pleasant experience. I open two terminal windows and place them one next to the other. I run `cargo lambda watch` in the right one, and `cargo lambda invoke -F events/flights.json` in the left.


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/e7gntefll8kz134hwm6s.png)

As you might imagine, I needed a few iterations before I got the `FlightData` deserialization properly defined. With `cargo lambda` iterating is super smooth - in the watch mode lambda is rebuilt automatically on save and can be invoked right away. 

# Summary

In the current post, I prepared a local environment for building a lambda function that consumes data from OpenSearch. This is a nice starting point.

As the next steps, I will implement more real-life queries, write an indexer function to populate data, and finally, deploy the whole solution to AWS. Bear with me if you are interested in this topic.