Unable to detect if a watch is active

[maxweisel]

This is the same bug that occurred in the Go client library here: kubernetes/kubernetes#65012

Essentially if you create a watch stream and the connection is severed, the stream is unaware and believes it is open. It will stay open forever and just stop reporting events instead of trying to reconnect. This occurs anytime our k8s control plane is updated. Ideally it would detect the broken connection and reconnect to another replica or at least mark the connection closed.

It seems the solution is to occasionally http2 ping the connection to ensure that it's still open, and close if the ping is not received.

I've dug into the library a little and it seems watch.js uses request which is now deprecated. The watch class does allow specifying my own request implementation in the constructor, which I imagine I could use to add the ping feature, but I imagine this is a bug that will impact everyone who uses the watch event, so I figured I'd file an issue.

Happy to contribute a patch that includes this functionality, I'd love some guidance on what the preferred direction should be. Is there a suitable replacement for request that I could implement while I'm working on this patch?

Max

[brendandburns]

Thanks, please see the discussion of alternatives to request here:

#414

The main problem with switching libraries is that all of this code is generated by the OpenAPI Generator project:
https://github.com/OpenAPITools/openapi-generator

We could change generators, but that would also result in significant breaking API changes in the library. I think that the best option is to fork the request generator in OpenAPI Generator (typescript-node) and switch it to a different HTTP library while maintaining the same generated API signatures.

This is a pretty significant undertaking though, and for now no one has taken it on. If there is interest, I'm happy to advise on what needs to be done. I don't have the cycles for it personally, but I can act as a guide.

Regarding hanging watches, I think the best answer is to set the socket timeout pretty low (e.g. 1 minute) which should cause the TCP connection to reset anyway.

[burn2delete]

@brendandburns I would be willing to attempt migrating to a new generator.

[entropitor]

@maxweisel Thank you for discovering this! We have been plagued by this and I wasn't sure what the problem was

@jhagestedt It seems 1b313ce might be the culprit. Why is the timeout set to 0?

(Although I'm using 0.11.1 and I also seem to have the problem)

[entropitor]

I also wonder whether https://github.com/kubernetes-client/javascript/blame/master/src/cache.ts#L86 should maybe be before calling the error handlers so that if the error handlers immediately restart, it might end up in a stopped state because we first started (stopped = false) and then the error handler sets it back to "stopped"

[maxweisel]

@brendanburns fwiw, I don't think setting the socket timeout to one minute is an acceptable solution here. Our control plane runs a very large number of pods and when the stream initially connects it will send the entire list of all running pods and their current states.

Our application fields requests by spinning up pods to do work. If our watch stream goes down, our service grinds to a halt as it can't confirmation that work for the request has started. For our service to run properly, I need to detect that the stream has gone down and reconnect within 5-10 seconds. I could set the timeout to 5 seconds, but that means we'll be pulling the entire list of pods every 5 seconds which is also too much data to comb through.

I believe the correct solution here is to support http2 ping and ping every X seconds on the stream which is what the kubernetes go client is doing. The pings are cheap and the stream won't reset if the connection is active.

[brendandburns]

@flyboarder

I'd welcome the help.

The generator is here:

https://github.com/OpenAPITools/openapi-generator/tree/master/modules/openapi-generator/src/main/resources/typescript-node

I don't know if it is easier to clone the generator into a new directory and create a whole new generator, or to introduce a flag and just patch around the calls to request with calls to got or someother modern HTTP library.

Please take a look and let me know what you think.

Let's continue the discussion on #414 since that is the relevant issue and I don't want to hijack this issue.

[brendandburns]

@entropitor that code hasn't been released in any published versions yet, so I don't think it is causing this issue. (we should make it configurable though)

[brendandburns]

@maxweisel if you want to re-implement the watch code to use http 2 I'd be glad to take a PR.

fwiw, though, the Kubernetes API was not intended to be an event bus. I think you will find that using a true event bus like Kafka is a more resilient architecture, alternately, you could register an admission controller that gets called whenever a Pod is created, which would get you out of the watch business entirely.

[maxweisel]

Thanks for the notes. My use case is more complex than I let on and an admission controller is not a reasonable solution for me unfortunately :(

I'd be happy to throw an http2 ping PR together. A quick glance at request looks like I can't easily use http2 there. got looks great, but it uses http2-wrapper which makes accessing the http2session object needed to send a ping a little difficult.

I'm tempted to start with my own RequestInterface object that uses the raw http2 nodejs API, but I don't know if that's something you'd want upstream as it would mean watch would use a different request mechanism than all other requests in the library.

I'm going to experiment a bit and will report back.

[brendandburns]

@maxweisel thanks for being willing to take on a PR. I don't mind if we use a totally separate HTTP2 API for watch, however we should definitely make sure that we can reuse all of the authentication and configuration code.

e.g. if you wanted to re-implement watch.ts with a native node HTTP2 I think that's fine, as long as the interfaces/apis defined in that file stay the same.

[entropitor]

What is the best way to move forward?

@brendanburns Do you see a way of fixing this bug this without having to go to HTTP2?

[geoffpatehutch]

I found this issue as we're seeing the same problem. Sorry this isn't a full PR fix, but as a slightly fudged workaround, we fixed the issue by wrapping the Watch and Informer in our own class and injecting an HTTP2 request (rather than using the default request implementation) which we can keep open and check the status of the underlying stream. Hope this example code helps someone:

`
const http2 = require("http2");
const k8s = require("@kubernetes/client-node");

function KubernetesPodCache(namespace) {
this.informer = null;
this.namespace = namespace;

const onError = this.onError;
const onAdd = this.onAdd;
const onUpdate = this.onUpdate;
const onDelete = this.onDelete;

const that = this;
const k8sConfig = new k8s.KubeConfig();
k8sConfig.loadFromDefault();

const k8sApi = k8sConfig.makeApiClient(k8s.CoreV1Api);

const listFn = () => { 
    return k8sApi.listNamespacedPod(this.namespace);
};

const http2Request = {
    webRequest(opts, callback) {
        const connectionOptions = {};
        k8sConfig.applyToRequest(connectionOptions);

        const url = new URL(opts.uri);
        const host = `${url.protocol}//${url.host}`;

        const http2ClientSession = http2.connect(host, { ca: connectionOptions.ca });
        http2ClientSession.on("error", (err) => {
            console.error(err);
        });

        let path = `/api/v1/namespaces/${that.namespace}/pods?watch=true`;
        if (opts && opts.qs && opts.qs.resourceVersion) {
            path += `&resourceVersion=${opts.qs.resourceVersion}`;
        }

        const requestHeaders = { ":method": "GET", ":path": path }; 
        requestHeaders.Authorization = connectionOptions.headers.Authorization;

        const requestOptions = {
            "endStream": false
        };
            
        const http2Stream = http2ClientSession.request(requestHeaders, requestOptions);
        let count = 0;
        const pingInterval = setInterval(() => {
            let payload = count.toString().padStart(8, "0");
            payload = payload.slice(payload.length - 8);
            http2ClientSession.ping(Buffer.from(payload), (error, duration, payload) => {
                if ((error || http2Stream.closed) && that.informer && !that.informer.stopped) {
                    console.log(`Stopping Informer and clearing listeners and ping interval... error: ${error}, http2Stream.closed: ${http2Stream.closed}`);
                    that.informer.off("error", onError);
                    that.informer.off("add", onAdd);
                    that.informer.off("update", onUpdate);
                    that.informer.off("delete", onDelete);
                    that.informer.stop();
                    clearInterval(pingInterval);
                    http2ClientSession.close();
                    if (!http2Stream.closed) {
                        http2Stream.emit("close");
                    }

                    console.log("...Informer Stopped. Will auto restart soon");
                }

                count++;
            });
        }, 2000);
        http2Stream.on("end", () => {
            http2Stream.close();
        });
        
        return http2Stream;
    }
};

const watch = new k8s.Watch(k8sConfig, http2Request);
this.informer = new k8s.ListWatch(`/api/v1/namespaces/${this.namespace}/pods`, watch, listFn, false);

this.start(onError, onAdd, onUpdate, onDelete);

setInterval(() => {
    if (this.informer && this.informer.stopped) {
        console.log("Detected Informer stopped - restarting...");
        this.start(onError, onAdd, onUpdate, onDelete);
        console.log("...Informer started.");
    }

}, 30000);

}

KubernetesPodCache.prototype.start = function (onError, onAdd, onUpdate, onDelete) {
this.informer.start();
this.informer.on("error", onError);
this.informer.on("add", onAdd);
this.informer.on("update", onUpdate);
this.informer.on("delete", onDelete);
}

KubernetesPodCache.prototype.onError = function (error) {
console.error(error);
}

KubernetesPodCache.prototype.onAdd = function (obj) {
console.log("add " + obj.metadata.name + ", " + obj.status.phase);
};

KubernetesPodCache.prototype.onUpdate = function (obj) {
console.log("update " + obj.metadata.name + ", " + obj.status.phase);
};

KubernetesPodCache.prototype.onDelete = function (obj) {
console.log("delete " + obj.metadata.name + ", " + obj.status.phase);
};
`

[arsnyder16]

@geoffpatehutch Have you seen any issues with the code above? I am running into a similar problem where the watch connection closed but the library still thinks its open.

@brendandburns Is there any thought to adjusting the implementation simply for watch?

[arsnyder16]

@brendandburns I took @geoffpatehutch idea here and made a more general purpose drop in replacement for the request implementation for watches, i am goign to test this out the next few days. I realize there has been work in master changing the signature of this so it no longer takes a callback. what i have below matches the latest release 0.12.1

const http2 = require('http2');

const queryString = require('query-string');


class WatchRequest {
  webRequest(options, callback) {
    const {
      ca, headers, method, qs, uri
    } = options;
    const url = new URL(uri);
    const session = http2.connect(url, { ca });
    let ping = null;
    let error = '';
    session.on('error', err => error += err);
    session.on('close', () => {
      clearInterval(ping);
      callback(error);
    });
    const stream = session.request({ 
      ...headers,
      ':method': method, 
      ':path'  : `${url.pathname}?${queryString.stringify(qs)}`,
      'accept' : 'application/json'
    }, { 'endStream': false });
    stream.setEncoding('utf8');
    ping = setInterval(() => {
      session.ping(error => {
        if (error || stream.closed) {
          clearInterval(ping);
          if (!session.destroyed) {
            session.destroy(error || 'stream was closed');
          } else {
            console.error('session was already destroyed this is unexpected');
          }
        }
      });
    }, 2000);
    stream.on('error', () => {/* no opt this will allow session 'error' to be emitted instead of throwing an exception */});
    stream.on('close', () => {
      clearInterval(ping);
      session.close();
    });
    return stream;
  }
}

usage is just simply

const watch = new k8s.Watch(config, new WatchRequest());

[arsnyder16]

@brendandburns I was able to reproduce this fairly consistently in an AKS cluster, where the connection would seem to be open but in reality it was not

Curious if you experience the same:

https://github.com/arsnyder16/watch-rst-issue