From 03b3bf759a400aa223671cc647f2b2a2f9a42ab4 Mon Sep 17 00:00:00 2001 From: CritasWang Date: Fri, 31 Jul 2026 12:10:47 +0800 Subject: [PATCH] =?UTF-8?q?Add=20SessionPool.CheckHealthAsync=20connectivi?= =?UTF-8?q?ty=20probe=20=E6=96=B0=E5=A2=9E=20SessionPool.CheckHealthAsync?= =?UTF-8?q?=20=E8=BF=9E=E9=80=9A=E6=80=A7=E6=8E=A2=E9=92=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit IsOpen() is a lifecycle flag - it reports whether the caller has opened the pool and not yet closed it. Because the client keeps no heartbeat, it stays true after the server goes down, and users routinely mistake it for a health check, e.g. `if (pool.IsOpen()) return;` in a lazy-open guard, which then never rebuilds the pool. This adds the connectivity check that IsOpen() is mistaken for, without changing IsOpen() itself. IsOpen() 是生命周期标志，仅表示调用方是否已打开且尚未关闭连接池。由于客户端无心跳， 服务端断开后它仍为 true，用户普遍将其误当作健康检查（例如惰性打开守卫里的 `if (pool.IsOpen()) return;`），从而导致连接池永远不会被重建。本提交新增了 IsOpen() 被误认为具备的连通性检查能力，且不改变 IsOpen() 自身行为。 - Add SessionPool.CheckHealthAsync(CancellationToken), which issues one lightweight request on an idle pooled connection and returns a SessionPoolHealth snapshot carrying Status, AvailableClients, TotalPoolSize, FailedReconnections, Message and the underlying Error. 新增 SessionPool.CheckHealthAsync(CancellationToken)：在空闲连接上发送一次轻量请求， 返回包含状态、可用连接数、池容量、重连失败次数、说明和底层异常的 SessionPoolHealth 快照。 - Add SessionPoolHealthStatus with four outcomes: NotOpen (no network call is made), Healthy, Degraded (open but no idle connection to probe with - a saturation signal, not a server verdict) and Unhealthy (a connection was available but the server did not answer). 新增 SessionPoolHealthStatus 四种结果：NotOpen（不发起网络调用）、Healthy、 Degraded（池已打开但无空闲连接可用于探测，表示饱和而非服务端故障）、 Unhealthy（有可用连接但服务端未响应）。 - Add ConcurrentClientQueue.TryTake so the probe never blocks. A health endpoint must not queue behind application load, so when every client is busy the probe returns Degraded immediately instead of waiting. 新增 ConcurrentClientQueue.TryTake 使探针永不阻塞：健康检查端点不应排在业务负载之后， 因此所有连接繁忙时立即返回 Degraded 而不是等待。 - The borrowed connection is always returned to the pool and a failed probe does not close it, so the existing reconnect-on-use path is unchanged. 借出的连接总会归还池中，探测失败也不会关闭它，因此既有的按需重连路径保持不变。 - Forward CheckHealthAsync from TableSessionPool with identical semantics. TableSessionPool 以相同语义转发 CheckHealthAsync。 - Add tests for the NotOpen path, non-blocking TryTake semantics and the health snapshot; document the API in docs/SessionPool_Exception_Handling.md and docs/API.md. 补充 NotOpen 路径、TryTake 非阻塞语义与健康快照的测试；在相关文档中说明该 API。 --- docs/API.md | 3 +- docs/SessionPool_Exception_Handling.md | 78 +++++++++--- src/Apache.IoTDB/ConcurrentClientQueue.cs | 7 +- src/Apache.IoTDB/SessionPool.cs | 56 ++++++++- src/Apache.IoTDB/SessionPoolHealth.cs | 107 +++++++++++++++++ src/Apache.IoTDB/TableSessionPool.cs | 9 ++ .../SessionPoolHealthTests.cs | 112 ++++++++++++++++++ 7 files changed, 351 insertions(+), 21 deletions(-) create mode 100644 src/Apache.IoTDB/SessionPoolHealth.cs create mode 100644 tests/Apache.IoTDB.Tests/SessionPoolHealthTests.cs diff --git a/docs/API.md b/docs/API.md index 43fdd04..0b890b0 100644 --- a/docs/API.md +++ b/docs/API.md @@ -44,7 +44,8 @@ var tablet = | -------------- | ------------------------- | ------------------------ | ----------------------------- | | Open | bool | open session | session_pool.Open(false) | | Close | null | close session | session_pool.Close() | -| IsOpen | null | check if the pool was opened and not yet closed by the caller. It is a lifecycle flag, **not** a connectivity probe: it stays `true` after the server goes down, because the client keeps no heartbeat and reconnects on demand instead. | session_pool.IsOpen() | +| IsOpen | null | check whether the pool was opened and not yet closed by the caller. It is a lifecycle flag, **not** a connectivity probe: it stays `true` after the server goes down, because the client keeps no heartbeat and reconnects on demand instead. For connectivity use `CheckHealthAsync` | session_pool.IsOpen() | +| CheckHealthAsync | CancellationToken=default | probe server connectivity on an idle pooled connection; returns a `SessionPoolHealth` snapshot | await session_pool.CheckHealthAsync() | | OpenDebugMode | LoggingConfiguration=null | open debug mode | session_pool.OpenDebugMode() | | CloseDebugMode | null | close debug mode | session_pool.CloseDebugMode() | | SetTimeZone | string | set time zone | session_pool.GetTimeZone() | diff --git a/docs/SessionPool_Exception_Handling.md b/docs/SessionPool_Exception_Handling.md index 99039de..c9f0dee 100644 --- a/docs/SessionPool_Exception_Handling.md +++ b/docs/SessionPool_Exception_Handling.md @@ -54,24 +54,62 @@ catch (SessionPoolDepletedException ex) } ``` -## `IsOpen()` is a lifecycle flag, not a health check +## Checking connectivity: `CheckHealthAsync` vs `IsOpen` -`SessionPool.IsOpen()` reports whether **you** have opened the pool and not yet closed it. It is not a -connectivity probe: +`SessionPool.IsOpen()` reports whether **you** have opened the pool and not yet closed it. It is a +lifecycle flag, not a connectivity probe: - It becomes `true` after a successful `Open()` and only returns to `false` when you call `Close()`. - The client runs no heartbeat, so a server that goes down does **not** flip it back to `false`. Reconnection happens lazily, on the next operation. -This means the following common guard never re-opens the pool, because the flag stays `true` forever: +This makes the following common guard a trap - it short-circuits forever, so the pool is never rebuilt: ```csharp -// Anti-pattern: this short-circuits even while every connection is dead +// Anti-pattern: IsOpen() stays true even while every connection is dead if (_pool != null && _pool.IsOpen()) return; ``` -To reason about actual availability, use the health metrics below, or simply let an operation throw -`SessionPoolDepletedException` and handle it. +Use `CheckHealthAsync` when you need to know whether the server is actually reachable. It issues one +lightweight request on an idle pooled connection and returns a `SessionPoolHealth` snapshot: + +```csharp +var health = await sessionPool.CheckHealthAsync(); + +if (!health.IsHealthy) +{ + Console.WriteLine(health); // e.g. "Unhealthy: The server did not answer the probe: ..." + Console.WriteLine(health.Status); // NotOpen | Healthy | Degraded | Unhealthy + Console.WriteLine(health.Error); // the underlying exception, when Status is Unhealthy +} +``` + +### Status values + +| Status | Meaning | +| ----------- | ------------------------------------------------------------------------------------------- | +| `NotOpen` | The pool has not been opened yet, or has already been closed. No network call is attempted. | +| `Healthy` | The server answered the probe. | +| `Degraded` | The pool is open but no connection was idle to probe with. Says nothing about the server. | +| `Unhealthy` | A connection was available but the server did not answer. See `Error` for the cause. | + +### Behaviour worth knowing + +- **The probe never blocks.** If every client is busy, it returns `Degraded` immediately instead of + queueing behind ordinary work, so a health endpoint cannot be starved by application load. It is also + unaffected by the pool wait timeout described below. +- **The borrowed connection is always returned** to the pool, and a failed probe does not close it. The + regular reconnect-on-use path still applies to the next operation. +- **`Degraded` is not a server verdict.** Under high concurrency it simply means the pool was saturated at + that instant. Treat a persistent `Degraded` as a sizing signal, not an outage. +- **Bound the probe yourself if you need to.** Pass a cancellation token: + +```csharp +using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2)); +var health = await sessionPool.CheckHealthAsync(cts.Token); +``` + +`TableSessionPool` exposes the same `CheckHealthAsync` method with identical semantics. ## Pool Wait Timeout @@ -94,7 +132,9 @@ var sessionPool = new SessionPool.Builder() When the wait budget is exhausted, the operation throws `SessionPoolDepletedException` with the reason `Connection pool is empty and wait time out(...ms)`. Raise `SetPoolWaitTimeoutInMs` if your workload -legitimately queues behind long operations; lower it if you would rather fail fast and retry. +legitimately queues behind long operations; lower it if you would rather fail fast and retry. The budget is +a single deadline for the whole call: waiters woken by a returned connection that they lose the race for do +not restart it. > **Note:** before this setting existed, the wait budget was derived from the connection timeout and then > misinterpreted as seconds, which turned the 500 ms default into a ~41 minute block. If you are upgrading @@ -333,6 +373,8 @@ var sessionPool = new SessionPool.Builder() ### Health Check Implementation +The simplest health check delegates to the built-in probe and only adds your own policy on top: + ```csharp public class SessionPoolHealthCheck { @@ -343,26 +385,27 @@ public class SessionPoolHealthCheck _pool = pool; } - public HealthStatus CheckHealth() + public async Task CheckHealthAsync() { - var availableRatio = (double)_pool.AvailableClients / _pool.TotalPoolSize; + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2)); + var health = await _pool.CheckHealthAsync(cts.Token); - if (_pool.FailedReconnections > 10) + if (health.Status == SessionPoolHealthStatus.Unhealthy) { return new HealthStatus { Status = "Critical", - Message = $"High reconnection failures: {_pool.FailedReconnections}", + Message = health.Message, Recommendation = "Check IoTDB server availability" }; } - if (availableRatio < 0.25) + if (health.Status == SessionPoolHealthStatus.Degraded) { return new HealthStatus { Status = "Warning", - Message = $"Low available clients: {_pool.AvailableClients}/{_pool.TotalPoolSize}", + Message = $"No idle connection to probe ({health.AvailableClients}/{health.TotalPoolSize} available)", Recommendation = "Consider increasing pool size" }; } @@ -370,7 +413,7 @@ public class SessionPoolHealthCheck return new HealthStatus { Status = "Healthy", - Message = $"Pool healthy: {_pool.AvailableClients}/{_pool.TotalPoolSize} available" + Message = health.Message }; } } @@ -383,6 +426,9 @@ public class HealthStatus } ``` +If you would rather not issue a network call on every scrape, the metrics below can be sampled on their +own - just remember they describe the pool, not the server. + ### Metrics Collection for Monitoring Systems ```csharp @@ -559,7 +605,7 @@ public class ProductionSessionPoolManager The SessionPool exception handling and health monitoring features provide comprehensive tools for building robust IoTDB applications: - Use `SessionPoolDepletedException` to understand and react to pool issues -- Treat `IsOpen()` as a lifecycle flag, never as a connectivity check +- Use `CheckHealthAsync` for connectivity checks; `IsOpen()` is a lifecycle flag only - Tune `SetPoolWaitTimeoutInMs` separately from `SetConnectionTimeoutInMs` - Monitor `AvailableClients`, `TotalPoolSize`, and `FailedReconnections`; read `UnrealizedCapacity` as capacity not yet materialized rather than as an outage signal diff --git a/src/Apache.IoTDB/ConcurrentClientQueue.cs b/src/Apache.IoTDB/ConcurrentClientQueue.cs index f59b631..12bb087 100644 --- a/src/Apache.IoTDB/ConcurrentClientQueue.cs +++ b/src/Apache.IoTDB/ConcurrentClientQueue.cs @@ -58,7 +58,6 @@ public void Return(Client client) public void AddRef() => Interlocked.Increment(ref _ref); public int GetRef() => Volatile.Read(ref _ref); public void RemoveRef() => Interlocked.Decrement(ref _ref); - /// /// The maximum time, in milliseconds, that waits for a client to be /// returned to the pool before throwing. Defaults to 10000 (10 seconds). @@ -79,6 +78,12 @@ public int Timeout set => TimeoutInMs = value * 1000; } + /// + /// Attempts to take a client without ever blocking. Returns false when no client is idle. + /// Use this for probes and diagnostics, which must not queue behind ordinary work. + /// + public bool TryTake(out Client client) => ClientQueue.TryDequeue(out client); + public Client Take() { Client client = null; diff --git a/src/Apache.IoTDB/SessionPool.cs b/src/Apache.IoTDB/SessionPool.cs index fff1ea3..f5747d4 100644 --- a/src/Apache.IoTDB/SessionPool.cs +++ b/src/Apache.IoTDB/SessionPool.cs @@ -471,12 +471,62 @@ public async Task Reconnect(Client originalClient = null, CancellationTo /// /// This reflects the lifecycle of the pool object only - it is NOT a server-connectivity probe. /// The client performs no heartbeat, so a server going down does not flip this back to false; - /// it stays true until is called. To reason about connectivity, use - /// , and , - /// or simply let an operation throw . + /// it stays true until is called. Use when you + /// need to know whether the server is actually reachable, or read , + /// and for pool state. /// public bool IsOpen() => !_isClose; + /// + /// Probes server connectivity by issuing one lightweight request on an idle pooled connection, + /// and returns a snapshot of the result. This is the connectivity check that + /// is often mistaken for. + /// + /// + /// The probe never blocks waiting for a connection: if every client is busy, the call returns + /// immediately rather than queueing behind ordinary + /// work. The borrowed connection is always returned to the pool, and a failed probe does not close + /// it - the regular reconnect-on-use path still applies to the next operation. + /// Pass a cancellation token if you want to bound how long the probe may take. + /// + public async Task CheckHealthAsync(CancellationToken cancellationToken = default) + { + if (_isClose || _clients == null) + { + return new SessionPoolHealth(SessionPoolHealthStatus.NotOpen, 0, TotalPoolSize, FailedReconnections, + "The pool has not been opened yet, or it has already been closed."); + } + + if (!_clients.TryTake(out var client)) + { + return new SessionPoolHealth(SessionPoolHealthStatus.Degraded, 0, TotalPoolSize, FailedReconnections, + "No idle connection was available to probe. The pool is either saturated by concurrent work or has lost its connections."); + } + + SessionPoolHealthStatus status; + string message; + Exception error = null; + try + { + await client.ServiceClient.getTimeZoneAsync(client.SessionId, cancellationToken); + status = SessionPoolHealthStatus.Healthy; + message = "The server answered the probe."; + } + catch (Exception e) + { + status = SessionPoolHealthStatus.Unhealthy; + message = $"The server did not answer the probe: {e.Message}"; + error = e; + _logger?.LogWarning(e, "Health probe failed for session pool"); + } + finally + { + _clients.Add(client); + } + + return new SessionPoolHealth(status, AvailableClients, TotalPoolSize, FailedReconnections, message, error); + } + public async Task Close() { if (_isClose) diff --git a/src/Apache.IoTDB/SessionPoolHealth.cs b/src/Apache.IoTDB/SessionPoolHealth.cs new file mode 100644 index 0000000..691d3fd --- /dev/null +++ b/src/Apache.IoTDB/SessionPoolHealth.cs @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +using System; + +namespace Apache.IoTDB +{ + /// + /// Outcome of a probe. + /// + public enum SessionPoolHealthStatus + { + /// + /// The pool has not been opened yet, or it has already been closed. + /// + NotOpen, + + /// + /// The server answered the probe on a pooled connection. + /// + Healthy, + + /// + /// The pool is open but could not be probed, because no connection was idle at that moment. + /// This says nothing about the server: the pool may simply be saturated by concurrent work. + /// + Degraded, + + /// + /// A connection was available but the server did not answer the probe. + /// + Unhealthy + } + + /// + /// A point-in-time snapshot of pool connectivity, returned by . + /// Unlike - which only reports whether the caller has opened the pool - + /// this reflects whether the server actually answered just now. + /// + public class SessionPoolHealth + { + /// + /// The probe outcome. + /// + public SessionPoolHealthStatus Status { get; } + + /// + /// True only when is . + /// + public bool IsHealthy => Status == SessionPoolHealthStatus.Healthy; + + /// + /// Idle clients in the pool at the time the snapshot was taken. + /// + public int AvailableClients { get; } + + /// + /// Configured maximum capacity of the pool. + /// + public int TotalPoolSize { get; } + + /// + /// Cumulative tally of reconnection failures since the pool was opened. + /// + public int FailedReconnections { get; } + + /// + /// Human-readable explanation of . + /// + public string Message { get; } + + /// + /// The exception that caused an result, if any. + /// + public Exception Error { get; } + + public SessionPoolHealth(SessionPoolHealthStatus status, int availableClients, int totalPoolSize, + int failedReconnections, string message, Exception error = null) + { + Status = status; + AvailableClients = availableClients; + TotalPoolSize = totalPoolSize; + FailedReconnections = failedReconnections; + Message = message; + Error = error; + } + + public override string ToString() + => $"{Status}: {Message} (available {AvailableClients}/{TotalPoolSize}, failed reconnections {FailedReconnections})"; + } +} diff --git a/src/Apache.IoTDB/TableSessionPool.cs b/src/Apache.IoTDB/TableSessionPool.cs index 921aba1..3d39e0c 100644 --- a/src/Apache.IoTDB/TableSessionPool.cs +++ b/src/Apache.IoTDB/TableSessionPool.cs @@ -70,6 +70,15 @@ public void OpenDebugMode(Action configure) sessionPool.OpenDebugMode(configure); } + /// + /// Probes server connectivity on an idle pooled connection. See + /// for the semantics. + /// + public async Task CheckHealthAsync(CancellationToken cancellationToken = default) + { + return await sessionPool.CheckHealthAsync(cancellationToken); + } + public async Task Close() { await sessionPool.Close(); diff --git a/tests/Apache.IoTDB.Tests/SessionPoolHealthTests.cs b/tests/Apache.IoTDB.Tests/SessionPoolHealthTests.cs new file mode 100644 index 0000000..16a6089 --- /dev/null +++ b/tests/Apache.IoTDB.Tests/SessionPoolHealthTests.cs @@ -0,0 +1,112 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +using System; +using System.Diagnostics; +using System.Threading.Tasks; +using NUnit.Framework; + +namespace Apache.IoTDB.Tests +{ + [TestFixture] + public class SessionPoolHealthTests + { + private static Client NewStubClient() => new Client(null, 1L, 1L, null, null); + + private static SessionPool NewUnopenedPool() + => new SessionPool.Builder().SetHost("127.0.0.1").SetPort(6667).Build(); + + [Test] + public async Task CheckHealthAsync_PoolNeverOpened_ReportsNotOpen() + { + var health = await NewUnopenedPool().CheckHealthAsync(); + + Assert.That(health.Status, Is.EqualTo(SessionPoolHealthStatus.NotOpen)); + Assert.That(health.IsHealthy, Is.False); + Assert.That(health.Error, Is.Null); + } + + [Test] + public async Task CheckHealthAsync_PoolNeverOpened_ReturnsPromptlyWithoutTouchingTheNetwork() + { + // The probe must not attempt to connect, and must never block on the empty queue. + var stopwatch = Stopwatch.StartNew(); + var health = await NewUnopenedPool().CheckHealthAsync(); + stopwatch.Stop(); + + Assert.That(health.Status, Is.EqualTo(SessionPoolHealthStatus.NotOpen)); + Assert.That(stopwatch.ElapsedMilliseconds, Is.LessThan(2_000)); + } + + [Test] + public void IsOpen_StaysFalseUntilOpened() + { + // IsOpen is a lifecycle flag; CheckHealthAsync is the connectivity probe. + Assert.That(NewUnopenedPool().IsOpen(), Is.False); + } + + [Test] + public void TryTake_EmptyQueue_ReturnsFalseImmediately() + { + var queue = new ConcurrentClientQueue { TimeoutInMs = 30_000 }; + var stopwatch = Stopwatch.StartNew(); + + var taken = queue.TryTake(out var client); + + stopwatch.Stop(); + Assert.That(taken, Is.False); + Assert.That(client, Is.Null); + Assert.That(stopwatch.ElapsedMilliseconds, Is.LessThan(1_000), "TryTake must never wait on the queue."); + } + + [Test] + public void TryTake_ReturnsIdleClientWithoutRemovingOthers() + { + var queue = new ConcurrentClientQueue(); + var first = NewStubClient(); + var second = NewStubClient(); + queue.Add(first); + queue.Add(second); + + Assert.That(queue.TryTake(out var taken), Is.True); + Assert.That(taken, Is.SameAs(first)); + Assert.That(queue.ClientQueue.Count, Is.EqualTo(1)); + } + + [Test] + public void SessionPoolHealth_ToStringIncludesStatusAndCounters() + { + var health = new SessionPoolHealth(SessionPoolHealthStatus.Unhealthy, 3, 8, 2, "probe failed", new Exception("boom")); + + Assert.That(health.IsHealthy, Is.False); + Assert.That(health.ToString(), Does.Contain("Unhealthy")); + Assert.That(health.ToString(), Does.Contain("3/8")); + Assert.That(health.ToString(), Does.Contain("2")); + } + + [Test] + public void SessionPoolHealth_HealthyStatusSetsIsHealthy() + { + var health = new SessionPoolHealth(SessionPoolHealthStatus.Healthy, 8, 8, 0, "ok"); + + Assert.That(health.IsHealthy, Is.True); + Assert.That(health.Error, Is.Null); + } + } +}