diff --git a/plugins/in_kubernetes_events/kubernetes_events.c b/plugins/in_kubernetes_events/kubernetes_events.c index 080733480cd..3d996806b54 100644 --- a/plugins/in_kubernetes_events/kubernetes_events.c +++ b/plugins/in_kubernetes_events/kubernetes_events.c @@ -452,7 +452,9 @@ static int process_watched_event(struct k8s_events *ctx, char *buf_data, size_t msgpack_unpacked result; msgpack_object root; msgpack_object *item = NULL; + msgpack_object *metadata; flb_sds_t event_type = NULL; + uint64_t resource_version; /* unpack */ msgpack_unpacked_init(&result); @@ -481,6 +483,15 @@ static int process_watched_event(struct k8s_events *ctx, char *buf_data, size_t } ret = process_event_object(ctx, event_type, item); + if (ret == 0) { + metadata = record_get_field_ptr(item, "metadata"); + if (metadata != NULL && + record_get_field_uint64(metadata, "resourceVersion", &resource_version) == 0 && + resource_version > ctx->last_resource_version) { + flb_plg_debug(ctx->ins, "set last resourceVersion=%" PRIu64, resource_version); + ctx->last_resource_version = resource_version; + } + } msg_error: flb_sds_destroy(event_type); @@ -607,6 +618,9 @@ static struct flb_http_client *make_event_watch_api_request(struct k8s_events *c } flb_sds_printf(&url, "?watch=1&resourceVersion=%" PRIu64, max_resource_version); + if (ctx->watch_timeout > 0) { + flb_sds_printf(&url, "&timeoutSeconds=%d", ctx->watch_timeout); + } flb_plg_info(ctx->ins, "Requesting %s", url); c = flb_http_client(ctx->current_connection, FLB_HTTP_GET, url, NULL, 0, ctx->api_host, ctx->api_port, NULL, 0); @@ -870,6 +884,9 @@ static int check_and_init_stream(struct k8s_events *ctx) goto failure; } initialize_http_client(ctx->streaming_client, ctx); + if (ctx->watch_timeout > 0) { + flb_http_set_read_idle_timeout(ctx->streaming_client, ctx->watch_timeout); + } /* Watch will stream chunked json data, so we only send * the http request, then use flb_http_get_response_data @@ -1013,6 +1030,13 @@ static struct flb_config_map config_map[] = { "Set the polling interval for each channel (sub seconds)" }, + { + FLB_CONFIG_MAP_TIME, "kube_watch_timeout", DEFAULT_WATCH_TIMEOUT, + 0, FLB_TRUE, offsetof(struct k8s_events, watch_timeout), + "Set the maximum Kubernetes watch duration and client read idle timeout. " + "Set to 0 to disable the timeout" + }, + /* TLS: set debug 'level' */ { FLB_CONFIG_MAP_INT, "tls.debug", "0", diff --git a/plugins/in_kubernetes_events/kubernetes_events.h b/plugins/in_kubernetes_events/kubernetes_events.h index 4bc87001e88..937c71a56ee 100644 --- a/plugins/in_kubernetes_events/kubernetes_events.h +++ b/plugins/in_kubernetes_events/kubernetes_events.h @@ -28,6 +28,7 @@ #define DEFAULT_INTERVAL_SEC "0" #define DEFAULT_INTERVAL_NSEC "500000000" +#define DEFAULT_WATCH_TIMEOUT "10m" /* Filter context */ struct k8s_events { @@ -36,6 +37,7 @@ struct k8s_events { int interval_sec; /* interval collection time (Second) */ int interval_nsec; /* interval collection time (Nanosecond) */ int retention_time; /* retention time limit, default 1 hour */ + int watch_timeout; /* watch request timeout */ /* Configuration parameters */ char *api_host; diff --git a/tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py b/tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py new file mode 100644 index 00000000000..3a258cf5fde --- /dev/null +++ b/tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py @@ -0,0 +1,177 @@ +import contextlib +import http.server +import json +import os +import threading +import time + +from utils.data_utils import read_file +from utils.test_service import FluentBitTestService + + +EVENT_UID = "watch-event-uid" +RECOVERED_EVENT_UID = "post-recovery-event-uid" + + +def _event(resource_version, uid): + timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + return { + "metadata": { + "creationTimestamp": timestamp, + "resourceVersion": str(resource_version), + "uid": uid, + } + } + + +class _KubeApiServer(http.server.ThreadingHTTPServer): + daemon_threads = True + + def __init__(self, server_address, handler_class): + super().__init__(server_address, handler_class) + self.lock = threading.Lock() + self.stop_event = threading.Event() + self.list_requests = 0 + self.watch_requests = 0 + self.watch_paths = [] + self.event = _event(2, EVENT_UID) + self.recovered_event = _event(3, RECOVERED_EVENT_UID) + + +class _KubeApiHandler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_GET(self): + if "watch=1" in self.path: + with self.server.lock: + self.server.watch_requests += 1 + self.server.watch_paths.append(self.path) + watch_request = self.server.watch_requests + + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + if watch_request <= 2: + event = self.server.event + if watch_request == 2: + event = self.server.recovered_event + payload = ( + json.dumps({"type": "ADDED", "object": event}) + "\n" + ).encode("utf-8") + self.wfile.write(f"{len(payload):x}\r\n".encode("ascii")) + self.wfile.write(payload) + self.wfile.write(b"\r\n") + self.wfile.flush() + if watch_request == 2: + time.sleep(0.5) + self.wfile.write(b"0\r\n\r\n") + self.wfile.flush() + return + self.server.stop_event.wait(timeout=30) + try: + self.wfile.write(b"0\r\n\r\n") + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError): + pass + return + + with self.server.lock: + self.server.list_requests += 1 + list_request = self.server.list_requests + + payload = json.dumps( + { + "kind": "EventList", + "apiVersion": "v1", + "metadata": {"resourceVersion": str(min(list_request, 2))}, + "items": [self.server.event] if list_request > 1 else [], + } + ).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + self.wfile.flush() + + def log_message(self, fmt, *args): + return + + +@contextlib.contextmanager +def _run_kube_api_server(): + server = _KubeApiServer(("127.0.0.1", 0), _KubeApiHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + try: + yield server + finally: + server.stop_event.set() + server.shutdown() + server.server_close() + thread.join() + + +def _write_config(tmp_path, kube_api_port): + token_file = tmp_path / "token" + token_file.write_text("test-token", encoding="utf-8") + config_file = tmp_path / "kubernetes_events_watch_timeout.conf" + config_file.write_text( + "\n".join( + [ + "[SERVICE]", + " Flush 1", + " Grace 1", + " Log_Level info", + " HTTP_Server On", + " HTTP_Port ${FLUENT_BIT_HTTP_MONITORING_PORT}", + "", + "[INPUT]", + " Name kubernetes_events", + f" Kube_URL http://127.0.0.1:{kube_api_port}", + f" Kube_Token_File {token_file}", + " tls Off", + " Interval_Sec 5", + " Interval_NSec 0", + " Kube_Watch_Timeout 1s", + "", + "[OUTPUT]", + " Name stdout", + " Match *", + ] + ), + encoding="utf-8", + ) + return config_file + + +def test_kubernetes_events_reconnects_stalled_watch(tmp_path): + with _run_kube_api_server() as kube_api_server: + config_file = _write_config(tmp_path, kube_api_server.server_address[1]) + service = FluentBitTestService(os.fspath(config_file)) + service.start() + log_file = service.flb.log_file + + try: + service.wait_for_condition( + lambda: RECOVERED_EVENT_UID + if kube_api_server.watch_requests >= 2 + and RECOVERED_EVENT_UID in read_file(log_file) + and read_file(log_file).count("kubernetes stream disconnected") >= 2 + else None, + timeout=20, + interval=0.25, + description="an event from the reconnected Kubernetes watch", + ) + finally: + service.stop() + + with open(log_file, encoding="utf-8") as log: + log_text = log.read() + assert kube_api_server.list_requests >= 2 + assert kube_api_server.watch_requests >= 2 + assert all("timeoutSeconds=1" in path for path in kube_api_server.watch_paths) + assert log_text.count(EVENT_UID) == 1 + assert log_text.count(RECOVERED_EVENT_UID) == 1