`vector` sink does not work in k8s with dynamic IP addresses

[afoninsky]

Currently, vector has its own input/sink based on protobuf (https://vector.dev/docs/reference/sinks/vector/). I believe its expected purpose is to create complex topologies with delivery guarantees in cases when people don't want or can't use kafka. Sink connects to downstream vectors and streams logs/events in a native form.

Unfortunately, it does not work in flexible environments (like "kubernetes"). Example:

[vector producer]{vector sink} → [k8s service] → {vector input}[vector consumer]

Here, k8s service acts as a load-balancer providing ip-adresses of downstream vectors in mDNS format OR forwards to one of downstream vectors using round-robin (perhaps actual behavior differs, I didn't check documentation :)

When [vector consumer] is restarted (ip address has changed) - [vector producer] does not handle it and keeps sending messages to nowhere. If I restart [vector producer] it reconnects to [k8s service] and reaches new consumer.

Possible things to improve: 1) Fix healthcheck (for now, it does not handle absence of downstream consumer at all). 2) Implement automatic reconnection algorithm on health failure. 3) Add ability to customize such algorithm (for example, it will be possible to add more than one consumer and achieve high availability, scale consumption of resources) 4) Switch to GRPC (it already has some loadbalancing/retry/etc features out-of-the-box, also modern loadlabalncers [nginx,envoy,etc] able to understand grpc calls which gives more flexibility) 5) Make him understand mDNS (not sure if it's useful but in case if it's not very expensive to add additional flag ...)

p.s.: is it possible to switch "vector" input/sink from "best-effort" to "at-least-once" somehow? :)

[binarylogic]

Thanks @afoninsky for the detailed writeup. @ktff is the best person to weigh in here.

p.s.: is it possible to switch "vector" input/sink from "best-effort" to "at-least-once" somehow? :)

Yep :) It's on our 1.0 roadmap. #492 addresses this. Between #492 and #807 the vector to vector communication should be fairly reliable.

[ktff]

The vector sink has the capability to restart a connection, full DNS resolving and so on, and it will once it detects any error. The problem seems to be that the code assumes no error means it is connected. Which by default isn't true for TCP protocol. So using keepalive feature of TCP protocol should fix this issue.

@afoninsky Regarding healthchecks, they are a one time opt in check, for now at least. Although, as you said, there is a potential for a lot of things.

For now let's focus on the bug. I'll try to reproduce it and test the fix.

[ktff]

Noting that, vector/TCP sink are properly detecting disconnection if tested on a local machine.

[MOZGIII]

Technically, with Service in play, producer Pod is supposed to connect not directly to the consumer Pod, but to the Service itself. The Service has it's own IP address assigned, and perform the connection routing to a set of associated Endpoints. When consumer Pod is recreated, K8s will remove the old and obsolete Endpoint and add a new one.

If we reconnect properly upon receiving a socket write error - should work fine.

A note on gRPC - load balancing gRPC in k8s it quite difficult. The problem lies with it's long-living connections, we'll have the same set of issues as with other protocol. Unless, of course, we use a service mesh with per-Pod L7 reverse-proxy, like istio.

[jamtur01]

@MOZGIII Can you clarify how we should proceed here?

[MOZGIII]

We need to validate that vector sink can re-resolve the DNS addresses when the records change. A similar implementation is in some ways implemented for kafka.

[MOZGIII]

Since in our k8s integration we'll be deploying Vector aggregator role as a StatefulSet, and, therefore, with a headless service, we might want to implement the support for _SRV records lookup at the k8s environment.

[spencergilbert]

@JeanMertz I've started reviewing the PR for vector source/sink v2, but I don't think it's added any support for dynamic client side load balancing - is that right?

I think this issue is the largest function based blocker to the vector-aggregator work

[JeanMertz]

@JeanMertz I've started reviewing the PR for vector source/sink v2, but I don't think it's added any support for dynamic client side load balancing - is that right?

As far as I'm aware, there's no need for this at this moment.

Currently, we don't use long-lived (streaming) connections in the gRPC draft implementation, this is intentional, because of the issues @MOZGIII mentioned above w.r.t. load balancing inside Kubernetes (or any dynamic load balancing system).

Instead, the implementation is based on batched message delivery similar to the HTTP source/sink, except that it uses the Tonic gRPC library on top of the Tower HTTP library, and encodes the messages as Protobufs using the Prost library.

While streaming might (most likely will) improve performance, it was deemed unneeded at this point, and the performance of batched deliveries should be sufficient to solve our current use-case.

As for the network topology; the Vector client connects to a Kubernetes service sitting in front of the Vector server, which then handles all the required load balancing. This means that the connection between the client and the Kubernetes service is stable, and re-routing traffic to/from removed/added pods is handled dynamically by Kubernetes itself.

Client-side load balancing is something we could consider in the future, but — as far as I'm aware at this moment — is not a requirement to get a working set-up going inside Kubernetes.

Does this align with your thoughts on this issue?

[JeanMertz]

To make it more clear why this issue exists with the current vector sink, but won't be an issue with the gRPC based one:

The current Vector sink communicates by streaming events over TCP, not over HTTP, which means we have to handle dynamic connections ourselves. The new implementation leverages gRPC over HTTP, and does not stream events, which (should) resolves this issue.

[jszwedko]

@JeanMertz Is this resolved by the gRPC rewrite?

[alexandv]

Hi,

I tested this new vector version 2 source/sink to work with a k8s service and load balance requests. However, the TCP connection is still persistent so requests will not be properly balanced between multiple pods behind a k8s service. Is it possible to force a reconnection after every X messages for example or after every flush? HTTP sink/source has the same issue.

[jszwedko]

Hi @alexandv ! That's correct that Vector will re-use TCP connections. We recommend using an HTTP load balancer to route the requests. I see the issue you opened around maximum number of requests / duration for TCP connections to though. I could see supporting that as well.

[brablc]

I suspect sink IP change in docker swarm to be same/similar thing leading to eternal:

2024-05-31T22:58:02.501621Z ERROR sink{component_kind="sink" component_id=rabbitmq component_type=amqp}:request{request_id=600779}: vector_common::internal_event::service: Internal log [Service call failed. No retries or retries exhausted.] is being suppressed to avoid flooding.

As HTTP load balancing would be an overkill for small swarms, I added a healtcheck to detect changed IP condition:

        command: >
            "
            export RABBITMQ_URL='...';
            dockerize -wait tcp://rabbitmq:5672 -timeout 60s -wait-retry-interval 5s true || exit 1;
            getent hosts rabbitmq | sed -E 's/\\s\\b/ orig-/g' >> /etc/hosts;
            exec vector -c /etc/vector/vector.yaml
            "
        healthcheck:
            test: ["CMD-SHELL", "dockerize -wait http://localhost:8686/health -wait tcp://orig-rabbitmq:5672 -timeout 5s true"]
            interval: 10s
            timeout: 5s
            retries: 5
            start_period: 60s