Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 66 additions & 0 deletions docs/health-backend.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
# Aggregate health backend

`OAdsGetServerStats(hMgmt, json, len)` is an additive OpenADS C extension.
Use an `AdsMgConnect` management handle, not a data connection handle.
No existing ADS management structure or serialized snapshot format changes.

`*len` is the supplied capacity in bytes, including the NUL terminator. On a
successful query it becomes the required size including NUL. A NULL or short
buffer returns `AE_INSUFFICIENT_BUFFER` and writes no partial JSON. Allocate
headroom or retry when the next sample has grown. An invalid handle returns
`AE_INVALID_CONNECTION_HANDLE`; a NULL length pointer is an error.

Local management handles sample the calling process (`scope: local_process`).
Remote handles sample the server (`scope: server`) through the additive
`MgRequestKind::HealthJson` read request. `AdsMgConnect` accepts host:port and
`tcp://host:port`; management uses plain TCP, not TLS. Existing management
credentials are sent on each management connection. Do not treat a TLS-enabled
data server build as encrypted management transport.

The existing management handshake and credential policy remain in force.
Configured credentials are checked. Without configured credentials, the
existing loopback-only read-only management session may read health; it still
cannot run management mutators. No handshake means no health read. An older
server that rejects the new request returns `AE_FUNCTION_NOT_AVAILABLE`.
Transport failures return `AE_NO_CONNECTION`; later handshake refusal returns
`AE_ACCESS_DENIED`. The client bounds replies to 65,535 bytes and checks an
object envelope, not full JSON syntax validation.

## Schema 1

The JSON contains aggregates, never usernames, client addresses, SQL text,
credentials, or table/index path strings. Distinct path counts use exact
stored strings, not canonical filesystem identity.

- `users`, `connections`, `workareas`, `table_handles`, `index_bindings`,
`locks`: `current`, `max_used`, and `rejected` (unknown, null).
- `worker_threads`: current existing server thread telemetry; peak and rejects
are unknown, null. This is not a claim of process-wide OS thread counting.
- `distinct_table_paths`, `distinct_index_paths`, `uptime_seconds`,
`operations`, `logged_errors`, `rss_bytes`, `server_port`.
- `max_sessions` and `max_sessions_source`: resolved server admission cap and
its source (default, environment, INI, command line, or API override).
Explicit server zero means unlimited. Both fields are null for a local
process sample. The default remains 500.
- `packets_in`, `packets_out`, `bytes_in`, `bytes_out`, `disconnects`,
`partial_connects`: existing process telemetry, with wide integer output.
- `os_fd_count` and `os_fd_soft_limit`: Linux process measurements when
available; null on other platforms or when unavailable/unlimited.
- `parked_handles`: null. Parking is client-only state.
- `semantics`: descriptions carried in every sample.

Workareas mean server-open table handles, including handles parked at the
client. They are not Harbour `Select()` areas and not OS file descriptors.
Workarea/table-handle maxima retain open events, including opens that close
before a dashboard sample. Other maxima keep the existing telemetry semantics.
Maxima are process-global; samples from multiple embedded Server instances
share them. Resource counts retain existing 32-bit widths; byte counters and
other wide counters do not truncate to 32 bits. A snapshot is a best-effort
concurrent sample; the management query session itself may appear in counts.

## Scope

This backend introduces no Harbour wrapper, portal/dashboard consumer,
listener, web endpoint, lock algorithm, count-cache policy, or daemon safeguard
exemption. The C extension is exported through the existing shared-library
export paths, including the Windows x86 stdcall wrapper.
5 changes: 5 additions & 0 deletions include/openads/ace.h
Original file line number Diff line number Diff line change
Expand Up @@ -1565,6 +1565,11 @@ UNSIGNED32 ENTRYPOINT AdsGetDefaultConnection(ADSHANDLE* phConn);
// OpenADS extension: process-wide logging kill-switch (audit + arc
// traces). 0 = silent (production), non-zero = enabled (default).
UNSIGNED32 ENTRYPOINT OAdsSetLogging(UNSIGNED16 usOn);
// Aggregate health JSON from an AdsMgConnect handle (NOT a data connection).
// *pulLen is buffer capacity incl. NUL; out is required size incl. NUL.
// NULL/short buffer: AE_INSUFFICIENT_BUFFER, no partial JSON written.
UNSIGNED32 ENTRYPOINT OAdsGetServerStats(ADSHANDLE hMgmt, UNSIGNED8* pucJson,
UNSIGNED32* pulLen);
UNSIGNED32 ENTRYPOINT AdsSetEpoch (UNSIGNED16 usEpoch);
UNSIGNED32 ENTRYPOINT AdsSetExact (UNSIGNED16 bExact);
UNSIGNED32 ENTRYPOINT AdsSetExact22 (ADSHANDLE hObj,
Expand Down
1 change: 1 addition & 0 deletions src/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ add_library(openads_core STATIC
network/remote_index_nav.cpp
network/mutex_manager.cpp
mgmt/mg_collector.cpp
mgmt/mg_health.cpp
mgmt/error_log.cpp
network/mg_wire.cpp
sql_backend/uri.cpp
Expand Down
53 changes: 53 additions & 0 deletions src/abi/ace_exports.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ thread_local std::string correlation;
#include "mgmt/error_log.h"
#include "mgmt/mg_collector.h"
#include "mgmt/mg_stats.h"
#include "mgmt/mg_health.h"
#include "session/connection.h"
#include "session/handle_registry.h"
#include "util/log.h"
Expand Down Expand Up @@ -40173,6 +40174,10 @@ UNSIGNED32 ENTRYPOINT AdsMgConnect(UNSIGNED8* pucServer, UNSIGNED8* pucUser,
if (be.mg_user.size() > 256 || be.mg_password.size() > 4096) return openads::AE_LOGIN_FAILED;
std::string srv = pucServer
? reinterpret_cast<const char*>(pucServer) : "";
// Management accepts the ordinary TCP endpoint spelling as well as
// host:port. Do not strip tls://: this backend does not negotiate TLS.
if (srv.rfind("tcp://", 0) == 0 || srv.rfind("TCP://", 0) == 0)
srv.erase(0, 6);
// Strip leading / trailing UNC slashes ("\\\\host\\").
while (!srv.empty() && (srv.front() == '\\' || srv.front() == '/'))
srv.erase(srv.begin());
Expand Down Expand Up @@ -40230,6 +40235,54 @@ UNSIGNED32 ENTRYPOINT AdsMgConnect(UNSIGNED8* pucServer, UNSIGNED8* pucUser,
return openads::AE_SUCCESS;
}

UNSIGNED32 ENTRYPOINT OAdsGetServerStats(ADSHANDLE hMgmt, UNSIGNED8* json, UNSIGNED32* len) {
if (!len) return openads::AE_INTERNAL_ERROR;
MgBackend be;
{
std::lock_guard<std::mutex> lock(g_mg_mu);
const auto it = g_mg_handles.find(hMgmt);
if (it == g_mg_handles.end()) return openads::AE_INVALID_CONNECTION_HANDLE;
be = it->second; // concurrent disconnect cannot invalidate this copy
}
std::string text;
if (!be.remote) {
std::lock_guard<std::recursive_mutex> lock(state().mu);
text = openads::mgmt::health_json(collect_local_abi_snapshot(), OPENADS_VERSION_STR);
} else {
openads::network::network_init();
auto connected = openads::network::connect_tcp(be.host, be.port);
if (!connected) return openads::AE_NO_CONNECTION;
auto socket = connected.value();
if (!authenticate_mg_socket(socket, be)) {
openads::network::sock_close(socket);
return openads::AE_ACCESS_DENIED;
}
openads::network::Frame req;
req.opcode = openads::network::Opcode::MgRequest;
const auto body = openads::network::encode_mg_request(
openads::network::MgRequestKind::HealthJson, 0);
req.payload.assign(body.begin(), body.end());
if (!openads::network::write_frame(socket, req)) {
openads::network::sock_close(socket); return openads::AE_NO_CONNECTION;
}
auto reply = openads::network::read_frame(socket);
openads::network::sock_close(socket);
if (!reply) return openads::AE_NO_CONNECTION;
if (reply.value().opcode != openads::network::Opcode::MgReplyAck)
return openads::AE_FUNCTION_NOT_AVAILABLE;
text.assign(reply.value().payload.begin(), reply.value().payload.end());
// Bound the reply and require a JSON object envelope. This is not a full JSON parser.
if (text.empty() || text.size() > 65535 || text.front() != '{' || text.back() != '}')
return openads::AE_INTERNAL_ERROR;
}
const UNSIGNED32 needed = static_cast<UNSIGNED32>(text.size() + 1);
const UNSIGNED32 capacity = *len;
*len = needed;
if (!json || capacity < needed) return openads::AE_INSUFFICIENT_BUFFER;
std::memcpy(json, text.c_str(), needed);
return openads::AE_SUCCESS;
}

UNSIGNED32 ENTRYPOINT AdsMgDisconnect(ADSHANDLE hMgmt) {
arc2_trace("AdsMgDisconnect");
std::lock_guard<std::mutex> g(g_mg_mu);
Expand Down
10 changes: 10 additions & 0 deletions src/abi/ace_stdcall_x86.c
Original file line number Diff line number Diff line change
Expand Up @@ -6060,3 +6060,13 @@ __declspec(dllexport) UNSIGNED32 __stdcall OAdsSetLogging(UNSIGNED16 a0) {
arc_log_rc("OAdsSetLogging", rc, 1, (uintptr_t)a0);
return rc;
}

/* Additive health export; x86 Harbour imports the stdcall spelling. */
#define OAdsGetServerStats oadsimpl_OAdsGetServerStats
extern UNSIGNED32 ENTRYPOINT OAdsGetServerStats(ADSHANDLE, UNSIGNED8*, UNSIGNED32*);
#undef OAdsGetServerStats
#pragma comment(linker, "/alternatename:_oadsimpl_OAdsGetServerStats=_OAdsGetServerStats")
#pragma comment(linker, "/export:OAdsGetServerStats=_OAdsGetServerStats")
__declspec(dllexport) UNSIGNED32 __stdcall OAdsGetServerStats(ADSHANDLE a0, UNSIGNED8* a1, UNSIGNED32* a2) {
return oadsimpl_OAdsGetServerStats(a0, a1, a2);
}
68 changes: 68 additions & 0 deletions src/mgmt/mg_health.cpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
#include "mgmt/mg_health.h"
#include "platform/proc.h"
#include <algorithm>
#include <set>
#include <sstream>
#include <locale>

namespace openads::mgmt {
namespace {
std::string quote(const std::string& text) {
std::string out = "\"";
const char* hex = "0123456789abcdef";
for (char raw : text) {
const auto c = static_cast<unsigned char>(raw);
if (c == '"' || c == '\\') { out += '\\'; out += static_cast<char>(c); }
else if (c < 32) { out += "\\u00"; out += hex[c >> 4]; out += hex[c & 15]; }
else out += static_cast<char>(c);
}
return out + "\"";
}
}
std::string health_json(const MgSnapshot& s, const std::string& version) {
std::ostringstream out;
out.imbue(std::locale::classic());
out << "{\"schema_version\":1,\"version\":" << quote(version)
<< ",\"scope\":" << quote(s.server_type == 1 ? "server" : "local_process");
auto number = [&](const char* key, std::uint64_t n) { out << ",\"" << key << "\":" << n; };
auto usage = [&](const char* key, std::uint32_t now, std::uint32_t peak) {
out << ",\"" << key << "\":{\"current\":" << now
<< ",\"max_used\":" << std::max(now, peak) << ",\"rejected\":null}";
};
usage("users", s.users, s.max_users);
usage("connections", s.connections, s.max_connections);
usage("workareas", s.workareas, s.max_workareas);
usage("table_handles", s.tables, s.max_tables);
usage("index_bindings", s.indexes, s.max_indexes);
usage("locks", s.locks, s.max_locks);
out << ",\"worker_threads\":{\"current\":" << s.worker_threads
<< ",\"max_used\":null,\"rejected\":null}";
std::set<std::string> tables, indexes;
for (const auto& t : s.table_list) tables.insert(t.name);
for (const auto& i : s.index_list) indexes.insert(i.name);
number("distinct_table_paths", tables.size());
number("distinct_index_paths", indexes.size());
number("uptime_seconds", s.uptime_seconds);
number("operations", s.operations);
number("logged_errors", s.logged_errors);
number("rss_bytes", s.rss_bytes);
number("server_port", s.server_port);
out << ",\"max_sessions\":" << (s.max_sessions_source.empty() ? "null" : std::to_string(s.max_sessions))
<< ",\"max_sessions_source\":" << (s.max_sessions_source.empty() ? "null" : quote(s.max_sessions_source));
number("packets_in", s.packets_in); number("packets_out", s.packets_out);
number("bytes_in", s.bytes_in); number("bytes_out", s.bytes_out);
number("disconnects", s.disconnects); number("partial_connects", s.partial_connects);
const auto count = platform::process_fd_count(), limit = platform::process_fd_limit();
out << ",\"os_fd_count\":" << (count ? std::to_string(*count) : "null")
<< ",\"os_fd_soft_limit\":" << (limit ? std::to_string(*limit) : "null")
<< ",\"parked_handles\":null"
<< ",\"semantics\":{\"workareas\":\"server-open table handles, includes client-parked handles; not Harbour Select() areas\","
<< "\"users\":\"management user entries, not distinct people or application instances\","
<< "\"distinct_paths\":\"exact path strings, not canonical filesystem identities\","
<< "\"max_used\":\"workarea/table-handle maxima retain open events; other maxima keep existing telemetry semantics\","
<< "\"parked_handles\":\"client-only state, unavailable at server\","
<< "\"rejected\":\"not measured\","
<< "\"snapshot\":\"best-effort concurrent sample; management query session may be included\"}}";
return out.str();
}
}
7 changes: 7 additions & 0 deletions src/mgmt/mg_health.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
#pragma once
#include "mgmt/mg_snapshot.h"
#include <string>
namespace openads::mgmt {
// Aggregate-only JSON. No user names, paths, SQL text or credentials.
std::string health_json(const MgSnapshot& snap, const std::string& version);
}
2 changes: 2 additions & 0 deletions src/mgmt/mg_snapshot.h
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,8 @@ struct MgSnapshot {
std::uint16_t server_type = 0; // 0 = unknown/local
std::uint64_t rss_bytes = 0; // RSS of the reporting process
std::uint16_t server_port = 0; // listener port (0 = local)
std::uint32_t max_sessions = 0; // health-only, resolved daemon admission cap
std::string max_sessions_source; // empty when no daemon limit is known

// Cumulative / historical telemetry — captured from the reporting
// process's MgStats so it travels the wire with the live counts
Expand Down
1 change: 1 addition & 0 deletions src/network/mg_wire.h
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ enum class MgRequestKind : std::uint8_t {
KillUser = 0x02, // arg: u16 conn_no
ResetCommStats = 0x03,
DumpTables = 0x04,
HealthJson = 0x05, // additive aggregate-only health reply
};

// Request payload: [u8 kind][optional args].
Expand Down
22 changes: 20 additions & 2 deletions src/network/server.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,8 @@ mgmt::MgSnapshot Server::build_mg_snapshot() const {
snap.connections = static_cast<std::uint32_t>(sessions.size());
snap.server_type = 1; // 1 = remote server
snap.server_port = port_;
snap.max_sessions = max_sessions_;
snap.max_sessions_source = max_sessions_source_;
snap.rss_bytes = openads::platform::process_rss_bytes();

{
Expand Down Expand Up @@ -362,7 +364,11 @@ std::uint64_t Server::register_session(const SessionInfo& info) {
auto id = next_session_id_.fetch_add(1);
SessionInfo si = info;
si.id = id;
open_workareas_ += si.open_tables;
sessions_info_.emplace(id, std::move(si));
auto& stats = openads::mgmt::process_mg_stats();
openads::mgmt::MgStats::bump_max(stats.max_workareas, open_workareas_);
openads::mgmt::MgStats::bump_max(stats.max_tables, open_workareas_);
// M9.25 — raise the connection high-water mark under info_mu_.
openads::mgmt::MgStats::bump_max(
openads::mgmt::process_mg_stats().max_connections,
Expand All @@ -372,7 +378,11 @@ std::uint64_t Server::register_session(const SessionInfo& info) {

void Server::unregister_session(std::uint64_t id) {
std::lock_guard<std::mutex> lk(info_mu_);
sessions_info_.erase(id);
auto it = sessions_info_.find(id);
if (it != sessions_info_.end()) {
open_workareas_ -= it->second.open_tables;
sessions_info_.erase(it);
}
// M9.25 — count one disconnect per terminated session. This runs
// exactly once per session via session_loop's SessionGuard dtor.
openads::mgmt::process_mg_stats()
Expand Down Expand Up @@ -434,6 +444,7 @@ void Server::add_session_table(std::uint64_t id, std::int32_t delta,
auto it = sessions_info_.find(id);
if (it == sessions_info_.end()) return;
auto& n = it->second.open_tables;
const auto before = n;
if (delta < 0) {
std::uint32_t d = static_cast<std::uint32_t>(-delta);
n = (n > d) ? n - d : 0;
Expand All @@ -448,6 +459,12 @@ void Server::add_session_table(std::uint64_t id, std::int32_t delta,
it->second.open_table_names.push_back(table_name);
}
}
// Event-driven retained maximum, not a dashboard sampling maximum.
// Count logical session opens, including parked handles, not OS FDs.
open_workareas_ = open_workareas_ - before + n;
auto& stats = openads::mgmt::process_mg_stats();
openads::mgmt::MgStats::bump_max(stats.max_workareas, open_workareas_);
openads::mgmt::MgStats::bump_max(stats.max_tables, open_workareas_);
}

bool Server::try_register_open(std::uint64_t id,
Expand Down Expand Up @@ -549,9 +566,10 @@ util::Result<void> Server::start(const std::string& host,
// Resolve enterprise limits once at start. A test/embedder override wins;
// otherwise take the env-loaded EnterpriseConfig (OPENADS_SERVER_*).
const auto& ecfg = openads::sql_backend::enterprise_config();
max_sessions_ = (max_sessions_override_ != 0)
max_sessions_ = has_max_sessions_override_
? max_sessions_override_
: ecfg.server_max_sessions;
if (!has_max_sessions_override_) max_sessions_source_ = ecfg.server_max_sessions_source;
ListenerOptions opts;
opts.host = host;
opts.port = port;
Expand Down
9 changes: 8 additions & 1 deletion src/network/server.h
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,11 @@ class Server {
// the env-loaded EnterpriseConfig value; call before start(). Production
// reads OPENADS_SERVER_MAX_SESSIONS / openads.ini max_sessions; this
// exists mainly for tests.
void set_max_sessions(std::uint32_t n) { max_sessions_override_ = n; }
void set_max_sessions(std::uint32_t n, const std::string& source = "API override") {
max_sessions_override_ = n;
has_max_sessions_override_ = true;
max_sessions_source_ = source;
}
// listen() backlog override for the primary + extra listeners (0 = keep
// the env default, OPENADS_SERVER_BACKLOG / 256). Production config comes
// from --backlog / openads.ini backlog.
Expand Down Expand Up @@ -266,6 +270,8 @@ class Server {
// tests inject a small cap without touching the env-loaded singleton.
std::uint32_t max_sessions_ = 0;
std::uint32_t max_sessions_override_ = 0;
bool has_max_sessions_override_ = false;
std::string max_sessions_source_;
std::atomic<std::uint32_t> rejected_sessions_{0};
int backlog_override_ = 0;

Expand Down Expand Up @@ -298,6 +304,7 @@ class Server {
// sessions_snapshot().
mutable std::mutex info_mu_;
std::unordered_map<std::uint64_t, SessionInfo> sessions_info_;
std::uint32_t open_workareas_ = 0; // aggregate, protected by info_mu_
std::atomic<std::uint64_t> next_session_id_{1};

public:
Expand Down
Loading
Loading