Using Valkey Glide as a library within a Rust Lambda on AWS?

[RonaldLaBelle-Deckers]

Hello Valkey Glide team!

Would you be able to direct me to any documentation on using the inner Glide library in Rust lambda?
We work entirely in Rust on AWS, and if you point us to the correct methods & structs we can take it from there.

[shohamazon]

Hello! 🙂

We currently don't publish our Rust client as a crate. However, if you'd like to try Valkey Glide in Rust, you can use the following example code:
Valkey Glide Rust Benchmark.

You can build and use it locally. If you try it out and find it useful, we’d love to hear your feedback! Also, if you're interested in contributing to the development of the Rust client, let us know—we’d be happy to collaborate.

Let us know how it goes! 🚀

[RonaldLaBelle-Deckers]

Thanks for the direction! I'll take a look.

[avifenesh]

@RonaldLaBelle-Deckers did something come out of it?

[RonaldLaBelle-Deckers]

Hi @avifenesh ,

Unfortunately, I have not yet had a chance to test although the example indeed looks very promising. I was off-tasked by some exploratory around OpenTelemetry family of crates. I have an interest in testing this just finding the time has been tough. I will post findings when I have them.

[avifenesh]

Thanks! There's open telemetry development going in the core if you are interested.
Update when you have a plan, maybe we can partner on the development and create lean Rust client.
Opposed to the robust clients exist in the ecosystem, we can tailor something leaner and simpler, with strong focus on stability, reliability, performance, and important features.
While removing the options to choose between infinity of runtimes, tls packages, connection type, and so on and so forth.
Making it forkable for whom want to tailor something slightly different.

[avifenesh]

@kodecreer

[AfterThunk]

I'm new to AWS Lambdas and Valkey, but decided to give this a go because we've been looking into elasticache.
This is what I did:

Prereqs: Setup Cargo Lambda

https://www.cargo-lambda.info/

• Following the instructions for downloading cargo-lambda
• Create a new cargo-lambda project
• I'm on Windows, which means I had to have a cross compilation toolchain available too.
  I'll be honest, that's not something I know a ton about.
  Luckily, Strawberry Perl bundles a cross comp toolchain, and I had to have that installed for some other AWS SDKs anyway.
• Rust cross compilation also seems to require zig, so I downloaded that and made sure it was on my path.

Adding glide-core to the project

• Add glide-core and redis to the Cargo.toml

glide-core = { git = "https://github.com/valkey-io/valkey-glide.git", rev = "875088331f4fce35b4779ab37092de6399e9464e" }
redis = { git = "https://github.com/valkey-io/valkey-glide.git", rev = "875088331f4fce35b4779ab37092de6399e9464e" }

Compilation Errors

error: environment variable `GLIDE_NAME` not defined at compile time
    --> .cargo\git\checkouts\valkey-glide-7602178f3dd69e53\8750883\glide-core\redis-rs\redis\src\connection.rs:1011:14
     |
1011 |         .arg(std::env!("GLIDE_NAME"))
     |              ^^^^^^^^^^^^^^^^^^^^^^^
     |
     = help: use `std::env::var("GLIDE_NAME")` to read the variable at run time
     = note: this error originates in the macro `std::env` (in Nightly builds, run with -Z macro-backtrace for more info)

error: environment variable `GLIDE_VERSION` not defined at compile time
    --> .cargo\git\checkouts\valkey-glide-7602178f3dd69e53\8750883\glide-core\redis-rs\redis\src\connection.rs:1017:14
     |
1017 |         .arg(std::env!("GLIDE_VERSION"))
     |              ^^^^^^^^^^^^^^^^^^^^^^^^^^
     |
     = help: use `std::env::var("GLIDE_VERSION")` to read the variable at run time
     = note: this error originates in the macro `std::env` (in Nightly builds, run with -Z macro-backtrace for more info)

Had to add a cargo.toml file under .cargo with this:

[env]
GLIDE_NAME = "Glide" # This should be overwritten by each wrapper library.
GLIDE_VERSION = "unknown"

Adding a glide-core client to Lambda Function

I set the glade client up so that it can be shared among multiple requests:

// main.rs
use lambda_http::{run, service_fn, tracing, Error};
mod http_handler;

struct SharedResources {
    pub(crate) gamelift_client: aws_sdk_gamelift::Client,
    pub(crate) dynamodb_client: aws_sdk_dynamodb::Client,
    pub(crate) glide_client: glide_core::client::Client
}

async fn make_glide_client() -> glide_core::client::Client {
    // These variables are set with the lambda deployments.
    // https://docs.aws.amazon.com/lambda/latest/dg/configuration-envvars.html#:~:text=To%20set%20environment%20variables%20in,Under%20Environment%20variables%2C%20choose%20Edit.
    let address_info = glide_core::client::NodeAddress {
        host: std::env::var("GLIDE_HOST_IP").map_err(Box::new).unwrap(),
        port: std::env::var("GLIDE_HOST_PORT").map_err(Box::new).unwrap().parse().map_err(Box::new).unwrap()
    };

    let connection_request = glide_core::client::ConnectionRequest {
        addresses: vec![address_info],
        cluster_mode_enabled: true,
        request_timeout: std::env::var("GLIDE_REQUEST_TIMEOUT").ok().and_then(|v| v.parse::<u32>().ok()),
        tls_mode: Some(glide_core::client::TlsMode::SecureTls),
        ..Default::default()
    };

    glide_core::client::Client::new(connection_request, None).await.unwrap()
}

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

    let sdk_config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
    let shared_resources = SharedResources{
        gamelift_client: aws_sdk_gamelift::Client::new(&sdk_config),
        dynamodb_client: aws_sdk_dynamodb::Client::new(&sdk_config),
        glide_client: make_glide_client().await
    };

    let shared_resources_ref = &shared_resources;
    let handler = move |event: lambda_http::Request| async move {
        http_handler::function_handler(shared_resources_ref, event).await
    };

    run(service_fn(handler)).await
}

Sample Service Code (with redactions)

// http_handler.rs
#[derive(serde::Deserialize, serde::Serialize)]
struct UpdateStatusRequest {
	ticket: Ticket,
	... Other Serializable fields
}


#[derive(serde::Deserialize, serde::Serialize)]
struct UpdateStatusResult {}

async fn update_status(
    shared_resources: &crate::SharedResources,
    data: UpdateRequest
) -> Result<UpdateStatusResult, lambda_http::Error> {

    let mut cmd = redis::Cmd::new();
    cmd.arg("SET")
        .arg(data.ticket.ticket_id.as_ref().unwrap())
        .arg(serde_json::to_string(&data.ticket).map_err(Box::new)?);

    let _ = shared_resources.glide_client
        .clone()
        .send_command(&cmd, None)
        .await
        .map_err(Box::new)?;

    Ok(UpdateMatchmakingStatusResult {})
}

#[derive(serde::Deserialize, serde::Serialize)]
struct GetStatusRequest {
	ticket_id: String
}

#[derive(serde::Deserialize, serde::Serialize)]
struct GetStatusResponse {
	ticket: Ticket
}


async fn get_status(
    shared_resources: &crate::SharedResources,
    data: GetStatusRequest
) -> Result<GetStatusResult, lambda_http::Error> {

    let mut cmd = redis::Cmd::new();
    cmd.arg("GET")
        .arg(data.ticket_id);

    let result = shared_resources.glide_client
        .clone()
        .send_command(&cmd, None)
        .await
        .map_err(Box::new)?;

    match result {
        redis::Value::SimpleString(val) => Ok(GetStatusResult {
            ticket: serde_json::from_str(&val).map_err(Box::new)?
        }),
        _ => {
            Err(StringErr::boxed("Invalid result from Valkey!"))
        }
    }

}

Setting up a valkey cache in AWS

I'm sure there's more config that you can do, but I followed the instructions here:
https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/GettingStarted.serverless-valkey.step1.html

aws elasticache create-serverless-cache --serverless-cache-name CACHE_RESOURCE_NAME --engine valkey

You have to wait for the cache to be ready, and you can periodically check it with:

aws elasticache describe-serverless-caches --serverless-cache-name CACHE_RESOURCE_NAME

Once the cache is ready, you should be able to get its endpoint information.
I set this up using Environment Variables in my Lambdas.

Unfortunately, you can't test elasticache resources locally, because the endpoints are only accesible behind AWS VPC.

Runtime errors

Notice in the code above I have tracing::init_default_subscriber() commented out.
Inside logger_core/src/lib.rs there are a bunch of log wrapper functions defined via the create_log macro, and they all attempt to initialize logging.
By default, tracing only lets you set up one subscriber.

For libraries it's typically bad practice to setup logging yourself, specifically because you don't know what / how the main application is going to set it up.
The whole "logger_core" module should probably go away for a "real" rust client.
For any of the Language Client Drivers, there should just be a single init function for the driver, and it can setup logging if necessary, instead of doing it lazily like this.

Caveat

I haven't actually been able to test this fully yet.
Using cargo lambda watch it compiles and runs, but the connection always times out (due to the VPC issues).

I'm waiting to get IAM permissions so I can actually deploy the lambda, but for now everything is at least compiling.

[avifenesh]

That's pretty outstanding! Thanks for sharing! About the logger, yes, it is not fits for rust standalone implementation, I explain here the idea #3628 (comment)
But you can just set it to off, and use yours.
Please keep updating on how it is working!

[mo-amzn]

@AfterThunk
Thanks for sharing this! Really appreciate the detailed write-up.
GLIDE_NAME / GLIDE_VERSION compile-time errors come from how we use Rust's env!() macro, which requires the variables to be set at compile time. you can simply update them to be

GLIDE_NAME = "valkey-glide"
GLIDE_VERSION = "1.3.4" // Or the current version you're using

Please keep us posted once you're able to fully test the deployment — we’d love to hear how it goes.

[AfterThunk]

Update:

Changing Logging

• I forked this repo
• Modified the logging init function to do nothing.
• Used my repo + commit ID in my Cargo.toml
• If you're using the lambda_http crate's tracing::init_default_subscriber(), then you can change the log level by using the AWS_LAMBDA_LOG_LEVEL environment variable

Compilation

This page on cargo-lambda actually calls out the cross compilation issues I ran into:
https://www.cargo-lambda.info/guide/cross-compiling.html
See my earlier comment

Deployment

Deployment worked fine with cargo lambda deploy once I had the binary compiling, and the appropriate permissions.
This post on AWS Re:Post has info on the necessary permissions:

https://repost.aws/questions/QUxTLQWN2CTDmcyf1FxxQxyA/minimum-permissions-to-deploy-a-lambda-function

Connection Issues

By default, elasticache is only accessible via VPC.
Even when running the Lambda, I was still getting connection failure issues.
What I didn't realize is that Lambdas are not in/on a VPC by default.

I'm not sure if this is the correct way to handle this, but I ended up adding my lambda to the same VPC as elasticache.

• Go to the Lambda page in the AWS Console (Lambda > Functions > <MY_FUNCTION>)
• Go to Configuration > VPC > Edit
• You can figure out which VPC will be used by looking at your elasticache settings.
• This can take a few minutes to actually update.

Lambda Function URL

I needed my lambda to be publicly accesible, so I used the --http flag when creating it via cargo lambda new --http
I also had to go and add a Function URL on the lambda page:

• Go to the Lambda page in the AWS Console (Lambda > Function > <MY_FUNCTION>)
• There'll be a spot for Function URL.
• Follow the setup.
  • In my case, I wanted the lambda to be publicly available, so I opted for no IAM user.

Running

At this point, the lambda successfully spins up, I can reach it via the Function URL, and it works:

// main.rs

use lambda_http::{run, service_fn, tracing, Error, Body, Request, Response, tracing};
use serde::{Serialize, Deserialize}

struct SharedResources {
    pub(crate) glide_client: glide_core::client::Client
}

async fn make_glide_client() -> glide_core::client::Client {
    // These variables are set with the lambda deployments.
    // https://docs.aws.amazon.com/lambda/latest/dg/configuration-envvars.html#:~:text=To%20set%20environment%20variables%20in,Under%20Environment%20variables%2C%20choose%20Edit.
    let address_info = glide_core::client::NodeAddress {
        host: std::env::var("GLIDE_HOST_IP").map_err(Box::new).unwrap(),
        port: std::env::var("GLIDE_HOST_PORT").map_err(Box::new).unwrap().parse().map_err(Box::new).unwrap()
    };

    let connection_request = glide_core::client::ConnectionRequest {
        addresses: vec![address_info],
        cluster_mode_enabled: false,
        request_timeout: std::env::var("GLIDE_REQUEST_TIMEOUT").ok().and_then(|v| v.parse::<u32>().ok()),
        tls_mode: Some(glide_core::client::TlsMode::SecureTls),
        ..Default::default()
    };

    glide_core::client::Client::new(connection_request, None).await.unwrap()
}

type LambdaResult<T> = Result<T, lambda_http::Error>;

#[tokio::main]
async fn main() -> LambdaResult<()> {
    lambda_http::tracing::init_default_subscriber();

    let sdk_config = aws_config::load_defaults(aws_config::BehaviorVersion::l
        glide_client: make_glide_client().await
    };

    let shared_resources_ref = &shared_resources;
    let handler = move |event: lambda_http::Request| async move {
        function_handler(shared_resources_ref, event).await
    };

    lambda_http::run(lambda_http::service_fn(handler)).await
}


struct StringErr {
    value: String
}

impl StringErr {
    pub fn boxed(value: &str) -> Box<StringErr> {
        Box::new(StringErr {
            value: value.to_string()
        })
    }
}

impl std::fmt::Debug for StringErr {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.value)
    }
}

impl std::fmt::Display for StringErr {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.value)
    }
}

///////////////////////////////////////////////////////////////////////////////
/// Request / Result types
///////////////////////////////////////////////////////////////////////////////
#[derive(Deserialize, Debug)]
enum RequestType {
    GetStatus(GetStatusRequest),
    UpdateStatus(UpdateStatusRequest)
}

#[derive(Serialize, Deserialize, Debug)]
enum ResultType {
    GetStatus(GetStatusResult),
    UpdateStatus(UpdateStatusResult)
}

trait RequestHandler {
    async fn handle(self, shared_resources: &crate::SharedResources) -> LambdaResult<ResultType>;
}

impl RequestHandler for RequestType {
    async fn handle(self, shared_resources: &crate::SharedResources) -> LambdaResult<ResultType> {
        match self {
            Self::GetStatus(request) => request.handle(shared_resources).await,
            Self::UpdateStatus(request) => request.handle(shared_resources).await
        }
    }
}


///////////////////////////////////////////////////////////////////////////////
/// Update Ticket status
///////////////////////////////////////////////////////////////////////////////
#[derive(Deserialize, Debug)]
struct UpdateStatusRequest {
    ticket: Ticket
}

#[derive(Deserialize, Serialize, Debug)]
struct UpdateStatusResult {}

impl RequestHandler for UpdateStatusRequest {
    async fn handle(self, shared_resources: &crate::SharedResources) -> LambdaResult<ResultType> {
        let mut cmd = redis::Cmd::new();
        cmd.arg("SET")
            .arg(self.ticket.ticket_id.as_ref().unwrap())
            .arg(serde_json::to_string(&self.ticket).map_err(Box::new)?);

        let _ = shared_resources.glide_client
            .clone()
            .send_command(&cmd, None)
            .await
            .map_err(Box::new)?;

        Ok(ResultType::UpdateStatus(UpdateStatusResult {}))
    }
}


///////////////////////////////////////////////////////////////////////////////
/// Get Ticket Status
///////////////////////////////////////////////////////////////////////////////
#[derive(Deserialize, Debug)]
struct GetStatusRequest {
    ticket_id: String
}

#[derive(Deserialize, Serialize, Debug)]
struct GetStatusResult {
    ticket: Ticket
}

impl RequestHandler for GetStatusRequest {
    async fn handle(self, shared_resources: &crate::SharedResources) -> LambdaResult<ResultType> {
        let mut cmd = redis::Cmd::new();
        cmd.arg("GET")
            .arg(self.ticket_id);

        let result = shared_resources.glide_client
            .clone()
            .send_command(&cmd, None)
            .await
            .map_err(Box::new)?;

        let ticket : Ticket = match result {
            redis::Value::SimpleString(val) => serde_json::from_str(&val).map_err(Box::new)?,
            redis::Value::BulkString(val) => serde_json::from_slice(&val).map_err(Box::new)?,
            _ => return Err(StringErr::boxed("Invalid result from Valkey!"))
        };

        Ok(ResultType::GetStatus(GetStatusResult { ticket }))
    }
}


///////////////////////////////////////////////////////////////////////////////
/// Main Function Handler
///////////////////////////////////////////////////////////////////////////////
async fn function_handler(shared_resources: &crate::SharedResources, event: Request) -> LambdaResult<Response<Body>> {

    tracing::debug!("function_handler: invoked!");
    let request : RequestType = match event.body() {
        lambda_http::Body::Empty => Err(StringErr::boxed("Requests cannot be empty!"))?,
        lambda_http::Body::Text(val) => serde_json::from_str(&val)?,
        lambda_http::Body::Binary(val) => serde_json::from_slice(&val)?
    };

    tracing::debug!("function_handler: request: {event:?}");

    // TODO: Bailing with an error causes a 502 Bad Gateway to be received by the client.
    let result = request.handle(shared_resources).await?;

    tracing::debug!("function_handler: result: {result:?}");

    let resp = Response::builder()
        .status(200)
        .header("content-type", "application/json")
        .body(serde_json::to_string(&result).map_err(Box::new)?.into())
        .map_err(Box::new)?;

    Ok(resp)
}

[avifenesh]

@AfterThunk 👏🏽
That is super cool, thank you so much for sharing.
Would you consider contributing it somehow, to make it accessible for others? Maybe we can lift it up, clean the rust code internally, and create a crate?

[AfterThunk]

I can create a pull request, and add this under the examples directory with some docs, if that works.
I can add a separate pull request for disabling the log stuff based on a compilation flag, if that works for y'all

I'll try to get these up by end of the week.

[avifenesh]

Both are great. We could have the normal compilation path, and a feature for rust direct users that disable compile. You can add to example, and maybe to the wiki. @asafpamzn WDYT?

[AfterThunk]

Pull request created here:
2e1356c