implement the last subresources

[clux]

Subresources have a lot of special case logic, that is not easily derivable from k8s_openapi. So far we have implemented (see subresources.rs):

• [x] pods/eviction (impl in #393)
• [x] pods/log (already implemented, used as example herein)
• [x] pods/attach (hard - impl in #360)
• [x] pods/portforward (hard #229, sketch in #446)
• [x] pods/exec (hard #229 - impl in #360)
• [ ] {service,node,pod}/proxy - see #1161
• [ ] pods/binding (see https://docs.rs/k8s-openapi/latest/src/k8s_openapi/v1_26/api/core/v1/binding.rs.html#79-100 )
• [x] ephemeral containers - #1153
• [x] certificatesigningrequest/approval (easy - follow from #487 and see docs) - #773
• [x] TokenRequest - for client util (same as csr approach, easy) - done in #989
• [x] arbitrary subresources - probably port Request work from https://github.com/clux/kube-rs/pull/487#issuecomment-817103635 to API - #773

The general outline for the http based subresources:

We need a definition of what they are in subresources. The way to upgrade would be:

1. Implement subresource queries straight on Resource without restrictions (for now):

impl Resource {
    /// Get a pod logs
    pub fn log(&self, name: &str, lp: &LogParams) -> Result<http::Request<Vec<u8>>> {
        ...
        // Check https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/ for urls
    }
}

2. Make a marker trait for the subresource:

pub trait Loggable {}

impl<K> Api<K>
where
    K: Clone + DeserializeOwned + Meta + Loggable,
{
    pub async fn log(&self, name: &str, lp: &LogParams) -> Result<String> {
        let req = self.api.log(name, lp)?;
        Ok(self.client.request_text(req).await?)
    }

    pub async fn log_stream(&self, name: &str, lp: &LogParams) -> Result<impl Stream<Item = Result<Bytes>>> {
        let req = self.api.log(name, lp)?;
        Ok(self.client.request_text_stream(req).await?)
    }
}

3. Designate what resources implement this subresource:

impl Loggable for k8s_openapi::api::core::v1::Pod {}

Write one test case for one resource in examples (e.g. examples/log_openapi.rs).

EDIT (0.46.0):

Now this is not as difficult for the specialized requests requiring websockets (ws feature). This was discussed at length in #229, and finally implemented in #360. Implementors of the last subresources ought to look at that PR for help.

[koiuo]

Hi, I'd like to give this a try (I'm interested in support for exec mostly and would like to contribute what I can (which is not much at this point) to bring its support to kube-rs sooner).

I do understand, that this task is just about implementing subresources in rust code, not about handling them correctly. However one item is particularly confusing to me

1. Implement subresource queries straight on Resource without restrictions (for now)

I started with the logs example and came up with this for the exec

    pub fn exec(&self, name: &str, ep: &ExecParams) -> Result<http:Request<Vec<u8>>> {
        let base_url = self.make_url() + "/" + name + "/" + "exec?";
        let mut qp = url::form_urlencoded::Serializer::new(base_url);

        if let Some(container) = &lp.container {
            qp.append_pair("container", &container);
        }

        ...

        let req = http::Request::post(urlstr);
        req.body(vec![]).map_err(Error::HttpError)
    }

The issue is that exec can't be performed over http. It used to require SPDY (which is now deprecated) and now relies on websockets. Hence, an attempt to send a http request to this endpoint will yield 400 as response.

I wanted to clarify, whether you deliberately have chosen Result<http::Request<...>> as a return type to serve as a placeholder or you had something else in mind here.

If it's the former, should we maybe resort to Any to prevent folks using kube-rs from relying on these parts of API while full support is not there yet (you can't do much with Any I imagine)?

[clux]

The issue is that exec can't be performed over http. It used to require SPDY (which is now deprecated) and now relies on websockets. Hence, an attempt to send a http request to this endpoint will yield 400 as response.

I wanted to clarify, whether you deliberately have chosen Result<http::Request<...>> as a return type to serve as a placeholder or you had something else in mind here.

If it's the former, should we maybe resort to Any to prevent folks using kube-rs from relying on these parts of API while full support is not there yet (you can't do much with Any I imagine)?

Honestly, I wasn't aware of how much harder subresources like exec or port-forward` were going to be. We have a bit of an open discussion about it in #229, but we haven't gotten to a satisfying conclusion there yet.

Would assume that we need another return type for exec. Any doesn't sound very useful from an api reuse pov, nor from a user pov. It almost certainly needs a new transport dependency and a new signature specifically for the subresource.

That said, I haven't thought too much about this yet. If you want to help sketch out something or discuss on #229 that would be much appreciated.

[clux]

0.46.0 introduces the ws feature using async-tungstenite to pave the way for the last subresources listed above. Have updated the issue slightly. Implementations of the last ws-based subresources can look at #360 for an excellent PR that covers (among the websocket handling itself) examples for implementing pods/attach and pods/exec.

[kazk]

I'll do pods/portforward. kube should probably just provide AsyncRead + AsyncWrite for each port and not bind to local ports. Users should be able to do that if they want to.

I have a rough sketch that can do something similar to Python client's example:

// Portforward nginx container
let mut pf = pods.portforward("example", &[80]).await?; // Portforwarder
let mut ports = pf.ports().unwrap(); // Vec<Port>
let mut port = ports[0].stream().unwrap(); // impl AsyncRead + AsyncWrite + Unpin
port.write_all(b"GET / HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\nAccept: */*\r\n\r\n")
    .await?;
let mut rstream = tokio_util::io::ReaderStream::new(port);
if let Some(res) = rstream.next().await {
    match res {
        Ok(bytes) => {
            let response = std::str::from_utf8(&bytes[..]).unwrap();
            println!("{}", response);
            assert!(response.contains("Welcome to nginx!"));
        }
        Err(err) => eprintln!("{:?}", err),
    }
}

I'll open a draft PR later (it's too messy at the moment to show...) to discuss the design.

That should make kube a gold client except for "Proto encoding".

Gold Requirements

• Support exec, attach, port-forward calls (these are not normally supported out of the box from swagger-codegen)
• Proto encoding

I'm not sure what that means. Do we need to support Protobufs?

[clux]

Yeah, that sounds sensible. Listening sounds like an application concern. In an example we can maybe use TcpListener or something and direct it to the writable stream to show how to do the kubectl port-forward equivalent.

protobufs

Ugh. Ah, yeah, I am not sure that's really possible atm. The go api has protobuf codegen hints (see api/types.go) like:

// +optional
// +patchMergeKey=name
// +patchStrategy=merge
EphemeralContainers []EphemeralContainer `json:"ephemeralContainers,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,34,rep,name=ephemeralContainers"`

whereas the (huge) swagger codegen has the equivalent json for that part of the PodSpec:

        "ephemeralContainers": {
          "description": "...",
          "items": {
            "$ref": "#/definitions/io.k8s.api.core.v1.EphemeralContainer"
          },
          "type": "array",
          "x-kubernetes-patch-merge-key": "name",
          "x-kubernetes-patch-strategy": "merge"
        },

think the 34 there would be required, but haven't looked at this much. Arnavion might be able to say for sure whether it's possible, I can ping him on k8s-openapi.

[kazk]

The official JavaScript client is Gold and just have something like this (only deserializing):

export class ProtoClient {
    public readonly 'config': KubeConfig;
    public async get(msgType: any, requestPath: string): Promise<any> {
        // ...
        const req = http.request(options);
        const result = await new Promise<any>((resolve, reject) => {
            let data = '';
            req.on('data', (chunk) => {
                data = data + chunk;
            });
            req.on('end', () => {
                const obj = msgType.deserializeBinary(data);
                resolve(obj);
            });
            req.on('error', (err) => {
                reject(err);
            });
        });
        req.end();
        return result;
    }
}

https://github.com/kubernetes-client/javascript/blob/master/src/proto-client.ts

(BTW, it can't portforward more than one port. https://github.com/kubernetes-client/javascript/blob/a8d4f1c7bea2b27403d380e22e492b6680f80b31/src/portforward.ts#L19)

I guess the requirement for Gold is not well-defined? Python one seems more capable, but that's Silver.

---

When I searched "Kubernetes Proto Encoding", https://kubernetes.io/docs/reference/using-api/api-concepts/#protobuf-encoding came up.

[clux]

Yeah, I think the client grading document is a thing they they wrote and applied to the clients they generated themselves. It technically doesn't apply to us (because we don't have our client in that place they say it should be in), but it's a nice benchmark otherwise. I have opened an issue to continue protobuf discussions in #371 at any rate.

[chroche]

Naive question about pods/eviction, this feature doesn't seem to get much love at all. Is there a possible workaround / a direct way to create ak8s_openapi Eviction? Is there a reason why there seems to be no interest in this feaure?

[clux]

It's probably just that no one has asked for it yet. It actually looks very simple. I can have a go at adding it.

[clux]

Eviction handled in #393

[jmintb]

Can I have a go at implementing ephemeral container support?

[clux]

@jmintb Go for it!

[jmintb]

Alright so as far as I can tell we want to support replacing, updating and reading ephemeral containers for a pod. I have found the following operations in https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#podspec-v1-core. PATCH /api/v1/namespaces/{namespace}/pods/{name}/ephemeralcontainers GET /api/v1/namespaces/{namespace}/pods/{name}/ephemeralcontainers PUT /api/v1/namespaces/{namespace}/pods/{name}/ephemeralcontainers

This also matches what the JS client does.

Am I missing anything?

[clux]

That looks right to me!

One minor point; we probably have to put this behind an k8s_openapi::k8s_if_ge_1_25 guard when importing the structs since afaikt it only stabilised in 1.25.

[jmintb]

Awesome, that is my understanding as well.

[jmintb]

I noticed while writing tests for the ephemeral containers, that I can submit an invalid patch without an error occurring, should that be the case or have I broken something?

Maybe that is just how the JSON merge patch works.

Examples:

Bad case, where the patch is only the Pod spec and not the entire Pod object. I would expect an error here but get none.

let patch = serde_json::json!(
           {
           "ephemeralContainers": [
           {
               "name": "myephemeralcontainer",
               "image": "busybox:1.34.1",
               "command": ["sh", "-c", "sleep 20"],
           },
           ]
        });

        pod = pods
            .patch_ephemeral_containers(
                &pod.name_any(),
                &PatchParams::default(),
                &Patch::Merge(patch.clone()),
            )
            .await?;

Valid case with a partial Pod object.

let patch: Pod = serde_json::from_value(json!({
            "spec":{
            "ephemeralContainers": [
            {
                "name": "myephemeralcontainer",
                "image": "busybox:1.34.1",
                "command": ["sh", "-c", "sleep 20"],
            },
            ]
        }}))?;

        pod = pods
            .patch_ephemeral_containers(
                &pod.name_any(),
                &PatchParams::default(),
                &Patch::Merge(patch.clone()),
            )
            .await?;