Kube-rs: Use `tower-service` to build more robust clients

[LucioFranco]

Hi this project is fantastic!

I was curious if you'd be interested in possibly making the api use tower_service::Service intead of just using reqwest internally.

A couple big reasons this might be good is that you don't need to depend on reqwest directly which can be a heavy dep. You can let users easily configure how they might wanna do retries, timeouts, etc. It is easier to use a mock like tower-test to let users test their kube code.

Now tower is still getting worked on as you can currently see with the master branch but I know at least for some of my use cases I would love to have this support.

Example code that I have used before with the older tower-service but should work just the same with 0.3 is here https://github.com/LucioFranco/tower-consul/blob/master/src/lib.rs#L33

If you notice this entire crate doesn't depend on hyper but allows you to plug hyper in. Hyper 0.13 will also support Service out of the box.

I wrote this a bit quick but just wanted to start a discussion!

[clux]

Hey there, thank you! And thanks for all your work on tower, even though I haven't actually gotten to use it yet.

Short answer; Yes, _possibly_!

Being able to replace reqwest has certainly been asked requested before :slightly_smiling_face:

Long answer

The ideas that tower presents very much align with a bunch of issues we already have tracked (#34, #78, possibly helps with #66 since that's not even in the reqwest async ver afaikt), and we do need a better testing story herein (which we don't even have an issue for atm).

That said, embedding a bunch more traits herein certainly looks like it'll make things a lot more complicated, and I have a few questions if you have are able to answer:

1. Would this influence the server side parts of things? I.e. does it move this library closer to something that is better supported by tower-web? And would this hinder compatibility with embedding it within other web frameworks such as actix-web / rocket?

2. Would this work with streaming api calls without having a hard dependency on hyper or reqwest? Like, we need to embed the logic on how we deal with chunking, so presumably we need to bind to one http client anyway?

3. When is the tower release? :slightly_smiling_face: Not that I wanna add more pressure on this (going to be gone for about a month anyway), just curious what timelines you are working with.

[LucioFranco]

So for server side of things, Im not sure exactly what you are looking for around this but it shouldn't affect anything because in the end it all turns into a future.

For streaming calls, that would be abstracted via the http-body crate which you should use anyways as the generic trait bound around Request<B: http_body::Body>.

So tower-service is already released which should enable you to support all the use cases you need right now.

Also sean's latest blog post talks about tower service which should explain it a bit better https://seanmonstar.com/

I am hopping to write soon a blog post that should explain how I think these pieces should fit together. I would also be happy to submit a PR at some point in the future to add this if no one else does :D

[clux]

It makes a lot more sense now, thank you.

I would be happy to take a PR for this.

[kazk]

Hi @LucioFranco, I've been experimenting to swap reqwest with hyper (#376). I refactored all API requests (including WebSocket connection) to go through a single method that calls hyper::Client::request. Then started to experiment with tower and ended up with kube::Client with inner tower::Service :tada: However, I've been having trouble figuring out one last piece and I'd really appreciate your help. The PR is huge and messy (it's not meant to be reviewed/merged), so I'll try summarizing what I'm trying to achieve here.

I'm thinking of providing kube::Service based on hyper::Client for users to use by default (can be created from kube::Config) so that their workflow is not affected. It's hyper::Client with Request modified based on kube::Config:

let client: HyperClient<_, Body> = HyperClient::builder().build(connector);
let inner = ServiceBuilder::new()
    .map_request(move |r| set_cluster_url(r, &cluster_url))
    .map_request(move |r| set_default_headers(r, default_headers.clone()))
    .service(client);

I'd like to handle Authorization header similarly, but it'll be async (may need to refresh the token) and can fail (returns kube::Error). What is the recommended approach?

This is currently done in kube::Client by calling Authentication::to_header and modifying the request before calling the inner service.

let (mut parts, body) = request.into_parts();
if let Some(auth_header) = self.auth_header.to_header().await? {
    parts.headers.insert(http::header::AUTHORIZATION, auth_header);
}

Because of some restrictions to make this fit in kube, the type of service used by kube::Client at the moment is

Buffer<BoxService<Request<hyper::Body>, Response<hyper::Body>, hyper::Error>, Request<hyper::Body>>

Buffer is necessary for cheap clone and not using generics to avoid type parameters in kube::Client.

[clux]

I'm not good at this either; I would've hoped a .map_request with an async block could have worked, but if not, maybe it's possible to do a Layer for it, from what I have seen they should at least to handle errors and be async.

[kazk]

map_request with async block doesn't work because that makes Service<Future>.

Creating AuthLayer should work. We need AuthService with AuthFuture (a future that wraps Authentication::to_header(), inserting header, and calling inner service). (I'm not familiar with pinning and working with futures directly, so I need to play with them first.)

AuthLayer should be something that works similar to the following (based on the code you found), but without requiring Clone:

fn auth_layer<S>(
    auth: Authentication,
) -> impl Layer<
    S,
    Service = impl Service<
        Request<Body>,
        Response = Response<Body>,
        Error = BoxError,
        Future = impl Future<Output = Result<Response<Body>, BoxError>> + Send,
    > + Clone,
> + Clone
where
    S: 'static + Service<Request<Body>, Response = Response<Body>> + Clone + Send,
    S::Error: Send + Sync,
    BoxError: From<S::Error>,
    S::Future: Send,
{
    tower::layer::layer_fn(move |svc: S| {
        let auth = auth.clone();
        tower::service_fn(move |mut req: Request<Body>| {
            let auth = auth.clone();
            let mut svc = svc.clone();
            async move {
                if let Some(auth_header) = auth.to_header().await? {
                    req.headers_mut().insert(AUTHORIZATION, auth_header);
                }
                Ok(svc.call(req).await?)
            }
        })
    })
}

[kazk]

I don't think we can get rid of Clone, but hyper::Client is Clone so it works fine. Just pushed the initial implementation.