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); + } + } +}