2911 lines
115 KiB
PHP
2911 lines
115 KiB
PHP
<?php
|
|
|
|
namespace classes;
|
|
|
|
use mysqli;
|
|
use Predis\Client as PredisClient;
|
|
use RuntimeException;
|
|
use Throwable;
|
|
|
|
class replication_manager
|
|
{
|
|
private const KIND_DATABASE = 'database';
|
|
private const KIND_REDIS = 'redis';
|
|
private const MARIADB_SEED_BATCH_ROWS = 500;
|
|
private const MARIADB_SEED_STEP_SECONDS = 3;
|
|
private const MARIADB_SEED_FREEZE_TTL_SECONDS = 900;
|
|
private const MARIADB_SCHEMA_ONLY_TABLES = [
|
|
'logs',
|
|
'edge_gateway_log_entries',
|
|
'edge_gateway_audit_logs',
|
|
'edge_gateway_shell_sessions',
|
|
'replication_status_snapshots',
|
|
'replication_operations',
|
|
'replication_audit_logs',
|
|
];
|
|
|
|
public function __construct()
|
|
{
|
|
replication_schema_bootstrap::ensureTables();
|
|
}
|
|
|
|
public function summary(bool $refresh = false): array
|
|
{
|
|
$this->ensureEnvironmentPrimaryRows();
|
|
if ($refresh) {
|
|
$this->refreshStatuses();
|
|
}
|
|
|
|
$databaseHosts = $this->listHosts(self::KIND_DATABASE);
|
|
$redisHosts = $this->listHosts(self::KIND_REDIS);
|
|
|
|
return [
|
|
'generated_at' => date('c'),
|
|
'database' => [
|
|
'primary' => $this->publicHost($this->primaryHost(self::KIND_DATABASE)),
|
|
'hosts' => array_map(fn(array $host): array => $this->publicHost($host), $databaseHosts),
|
|
'replication' => $this->buildReplicationSummary(self::KIND_DATABASE, $databaseHosts),
|
|
],
|
|
'redis' => [
|
|
'primary' => $this->publicHost($this->primaryHost(self::KIND_REDIS)),
|
|
'hosts' => array_map(fn(array $host): array => $this->publicHost($host), $redisHosts),
|
|
'replication' => $this->buildReplicationSummary(self::KIND_REDIS, $redisHosts),
|
|
],
|
|
'write_freeze' => application_write_freeze::state(),
|
|
];
|
|
}
|
|
|
|
public function dependencyReplication(string $kind): array
|
|
{
|
|
$kind = self::normalizeKind($kind);
|
|
$this->ensureEnvironmentPrimaryRows();
|
|
return $this->buildReplicationSummary($kind, $this->listHosts($kind));
|
|
}
|
|
|
|
public function addHost(string $kind, array $input, ?int $actorUserId = null): array
|
|
{
|
|
$kind = self::normalizeKind($kind);
|
|
$host = $this->normalizeHostInput($kind, $input);
|
|
|
|
$this->execute(
|
|
"INSERT INTO replication_hosts (
|
|
kind, label, host, port, database_name, database_index, username, password_secret,
|
|
admin_username, admin_password_secret, replication_username, replication_password_secret,
|
|
role, status, ssl_mode, options_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'replica', 'unknown', ?, ?)",
|
|
'sssissssssssss',
|
|
[
|
|
$kind,
|
|
$host['label'],
|
|
$host['host'],
|
|
$host['port'],
|
|
$host['database_name'],
|
|
$host['database_index'],
|
|
$host['username'],
|
|
$host['password_secret'],
|
|
$host['admin_username'],
|
|
$host['admin_password_secret'],
|
|
$host['replication_username'],
|
|
$host['replication_password_secret'],
|
|
$host['ssl_mode'],
|
|
self::jsonEncode($host['options']),
|
|
]
|
|
);
|
|
|
|
$id = $this->insertId();
|
|
$this->audit($kind, $id, 'host_added', $actorUserId, 'info', [
|
|
'label' => $host['label'],
|
|
'host' => $host['host'],
|
|
'port' => $host['port'],
|
|
]);
|
|
$this->writeBootstrapSnapshot();
|
|
|
|
return $this->publicHost($this->getHost($kind, $id));
|
|
}
|
|
|
|
public function testHost(string $kind, int $id, ?int $actorUserId = null): array
|
|
{
|
|
$kind = self::normalizeKind($kind);
|
|
$host = $this->getHost($kind, $id);
|
|
|
|
$status = $kind === self::KIND_DATABASE
|
|
? $this->testDatabaseHost($host)
|
|
: $this->testRedisHost($host);
|
|
|
|
$this->storeStatus($host, $status);
|
|
$this->audit($kind, $id, 'host_tested', $actorUserId, $status['blockers'] === [] ? 'info' : 'warning', [
|
|
'status' => $status['status'],
|
|
'replication_percent' => $status['replication_percent'],
|
|
'blockers' => $status['blockers'],
|
|
]);
|
|
|
|
return [
|
|
'host' => $this->publicHost($this->getHost($kind, $id)),
|
|
'status' => $status,
|
|
];
|
|
}
|
|
|
|
public function testCredentials(string $kind, array $input): array
|
|
{
|
|
$kind = self::normalizeKind($kind);
|
|
$host = $this->transientHost($kind, $input);
|
|
$options = $this->decodeOptions($host);
|
|
if ($kind === self::KIND_DATABASE && !empty($options['allow_preseeded_replica'])) {
|
|
$host['connect_without_database'] = true;
|
|
}
|
|
$status = $kind === self::KIND_DATABASE
|
|
? $this->testDatabaseHost($host)
|
|
: $this->testRedisHost($host);
|
|
|
|
if ($kind === self::KIND_DATABASE
|
|
&& ($host['role'] ?? '') !== 'primary'
|
|
&& !empty($options['allow_preseeded_replica'])
|
|
&& $status['status'] !== 'down') {
|
|
try {
|
|
$target = $this->databaseConnection($host, true);
|
|
try {
|
|
$seedBlockers = $this->databaseReplicaSeedBlockers($this->primaryHost(self::KIND_DATABASE), $host, $target);
|
|
} finally {
|
|
$target->close();
|
|
}
|
|
} catch (Throwable $throwable) {
|
|
$seedBlockers = [$throwable->getMessage()];
|
|
}
|
|
|
|
if ($seedBlockers !== []) {
|
|
$status['blockers'] = array_values(array_unique(array_merge($status['blockers'], $seedBlockers)));
|
|
$status['status'] = 'degraded';
|
|
if ((float)$status['replication_percent'] >= 100.0) {
|
|
$status['replication_percent'] = 99.99;
|
|
}
|
|
}
|
|
}
|
|
|
|
return [
|
|
'ok' => $status['status'] === 'ok',
|
|
'host' => $this->publicHost($host),
|
|
'status' => $status,
|
|
];
|
|
}
|
|
|
|
public function provisionHost(string $kind, int $id, ?int $actorUserId = null): array
|
|
{
|
|
$kind = self::normalizeKind($kind);
|
|
$host = $this->getHost($kind, $id);
|
|
$operationId = $this->activeOperationId($kind, $id, 'provision')
|
|
?? $this->startOperation($kind, $id, 'provision', $actorUserId);
|
|
|
|
try {
|
|
$result = $kind === self::KIND_DATABASE
|
|
? $this->provisionDatabaseHost($host, $operationId)
|
|
: $this->provisionRedisHost($host);
|
|
|
|
if (($result['operation']['status'] ?? null) === 'running') {
|
|
$this->audit($kind, $id, 'host_provision_progress', $actorUserId, 'info', $result);
|
|
return $result;
|
|
}
|
|
|
|
$status = $result['ok'] ? 'completed' : 'blocked';
|
|
$this->finishOperation($operationId, $status, (float)($result['replication_percent'] ?? 0), $result['message'] ?? null, $result['blockers'] ?? []);
|
|
$this->audit($kind, $id, 'host_provisioned', $actorUserId, $result['ok'] ? 'info' : 'warning', $result);
|
|
return $result;
|
|
} catch (Throwable $throwable) {
|
|
$this->finishOperation($operationId, 'failed', 0, null, [$throwable->getMessage()]);
|
|
$this->audit($kind, $id, 'host_provision_failed', $actorUserId, 'error', ['error' => $throwable->getMessage()]);
|
|
throw $throwable;
|
|
}
|
|
}
|
|
|
|
public function promoteHost(string $kind, int $id, ?int $actorUserId = null): array
|
|
{
|
|
$kind = self::normalizeKind($kind);
|
|
$host = $this->getHost($kind, $id);
|
|
if (($host['role'] ?? '') === 'primary') {
|
|
return [
|
|
'ok' => true,
|
|
'message' => 'Host is already primary.',
|
|
'host' => $this->publicHost($host),
|
|
'blockers' => [],
|
|
];
|
|
}
|
|
|
|
$operationId = $this->startOperation($kind, $id, 'promote', $actorUserId);
|
|
$owner = 'replication-promote-' . $kind . '-' . $id . '-' . bin2hex(random_bytes(4));
|
|
$lockHandle = $this->acquirePromotionLock();
|
|
|
|
try {
|
|
application_write_freeze::freeze('Replication promotion in progress.', $owner, 600);
|
|
$result = $kind === self::KIND_DATABASE
|
|
? $this->promoteDatabaseHost($host)
|
|
: $this->promoteRedisHost($host);
|
|
|
|
$this->finishOperation($operationId, 'completed', 100, $result['message'] ?? null, []);
|
|
$this->audit($kind, $id, 'host_promoted', $actorUserId, 'critical', $result);
|
|
return $result;
|
|
} catch (Throwable $throwable) {
|
|
$this->finishOperation($operationId, 'failed', 0, null, [$throwable->getMessage()]);
|
|
$this->audit($kind, $id, 'host_promotion_failed', $actorUserId, 'error', ['error' => $throwable->getMessage()]);
|
|
throw $throwable;
|
|
} finally {
|
|
application_write_freeze::unfreeze($owner);
|
|
$this->releasePromotionLock($lockHandle);
|
|
}
|
|
}
|
|
|
|
public function removeHost(string $kind, int $id, ?int $actorUserId = null): array
|
|
{
|
|
$kind = self::normalizeKind($kind);
|
|
$host = $this->getHost($kind, $id);
|
|
if (!self::replicationHostCanBeRemoved($host)) {
|
|
if (($host['role'] ?? '') === 'primary') {
|
|
throw new RuntimeException('Primary hosts cannot be removed. Promote a healthy replica first.');
|
|
}
|
|
throw new RuntimeException('Only inactive prior hosts or unhealthy replicas can be removed.');
|
|
}
|
|
$this->execute(
|
|
"UPDATE replication_hosts SET deleted_at = NOW(), status = 'removed' WHERE id = ? AND kind = ?",
|
|
'is',
|
|
[$id, $kind]
|
|
);
|
|
$this->audit($kind, $id, 'host_removed', $actorUserId, 'warning', [
|
|
'label' => $host['label'] ?? '',
|
|
'host' => $host['host'] ?? '',
|
|
]);
|
|
$this->writeBootstrapSnapshot();
|
|
|
|
return [
|
|
'ok' => true,
|
|
'message' => 'Replication host removed.',
|
|
'id' => $id,
|
|
'kind' => $kind,
|
|
];
|
|
}
|
|
|
|
public static function replicationHostCanBeRemoved(array $host): bool
|
|
{
|
|
$role = (string)($host['role'] ?? '');
|
|
if ($role === 'primary') {
|
|
return false;
|
|
}
|
|
if ($role === 'inactive') {
|
|
return true;
|
|
}
|
|
|
|
$status = (string)($host['status'] ?? 'unknown');
|
|
return $role === 'replica' && in_array($status, ['degraded', 'down', 'unknown', 'not_configured'], true);
|
|
}
|
|
|
|
public static function composeTemplate(array $input): array
|
|
{
|
|
$kind = self::normalizeKind((string)($input['kind'] ?? self::KIND_DATABASE));
|
|
$role = self::normalizeComposeRole((string)($input['role'] ?? 'replica'));
|
|
|
|
return $kind === self::KIND_DATABASE
|
|
? self::databaseComposeTemplate($input, $role)
|
|
: self::redisComposeTemplate($input, $role);
|
|
}
|
|
|
|
public static function normalizeKind(string $kind): string
|
|
{
|
|
$kind = strtolower(trim($kind));
|
|
if (in_array($kind, ['database', 'databases', 'mysql', 'db'], true)) {
|
|
return self::KIND_DATABASE;
|
|
}
|
|
if ($kind === self::KIND_REDIS) {
|
|
return self::KIND_REDIS;
|
|
}
|
|
throw new RuntimeException('Unsupported replication kind.');
|
|
}
|
|
|
|
private static function databaseComposeTemplate(array $input, string $role): array
|
|
{
|
|
$serviceName = self::composeIdentifier($input['service_name'] ?? null, 'mariadb-' . $role);
|
|
$volumeName = self::composeIdentifier($input['volume_name'] ?? null, $serviceName . '-data');
|
|
$database = self::composeScalar($input['database'] ?? null, 'nnks_db');
|
|
$username = self::composeScalar($input['username'] ?? null, 'nnks_db_user');
|
|
$image = self::composeImage($input['image'] ?? null, 'mariadb:11');
|
|
$hostPort = self::boundedInt($input['host_port'] ?? null, $role === 'primary' ? 3306 : 3307, 1, 65535);
|
|
$serverId = self::boundedInt($input['server_id'] ?? null, $role === 'primary' ? 1 : 2, 1, 4294967295);
|
|
$rootPassword = self::composePassword($input['admin_password'] ?? null);
|
|
$applicationPassword = self::composePassword($input['password'] ?? null);
|
|
$replicationUsername = self::composeScalar($input['replication_username'] ?? null, 'replication');
|
|
$replicationPassword = self::composePassword($input['replication_password'] ?? null);
|
|
$primaryHost = self::composeScalar($input['primary_host'] ?? null, '<primary-host>');
|
|
$primaryPort = self::boundedInt($input['primary_port'] ?? null, 3306, 1, 65535);
|
|
|
|
$command = [
|
|
'mariadbd',
|
|
'--server-id=' . $serverId,
|
|
'--log-bin=/var/lib/mysql/mariadb-bin',
|
|
'--binlog-format=ROW',
|
|
'--gtid-strict-mode=ON',
|
|
'--expire-logs-days=7',
|
|
];
|
|
if ($role === 'replica') {
|
|
$command[] = '--read-only=ON';
|
|
foreach (self::MARIADB_SCHEMA_ONLY_TABLES as $tableName) {
|
|
$command[] = '--replicate-ignore-table=' . $database . '.' . $tableName;
|
|
}
|
|
}
|
|
|
|
$lines = [
|
|
'services:',
|
|
' ' . $serviceName . ':',
|
|
' image: ' . self::yamlQuote($image),
|
|
' restart: unless-stopped',
|
|
' environment:',
|
|
' MARIADB_ROOT_PASSWORD: "${MARIADB_ROOT_PASSWORD:?set MARIADB_ROOT_PASSWORD}"',
|
|
' MARIADB_DATABASE: ' . self::yamlQuote($database),
|
|
' MARIADB_USER: ' . self::yamlQuote($username),
|
|
' MARIADB_PASSWORD: "${MARIADB_PASSWORD:?set MARIADB_PASSWORD}"',
|
|
' command:',
|
|
];
|
|
|
|
foreach ($command as $argument) {
|
|
$lines[] = ' - ' . self::yamlQuote($argument);
|
|
}
|
|
|
|
$lines = array_merge($lines, [
|
|
' volumes:',
|
|
' - ' . $volumeName . ':/var/lib/mysql',
|
|
' ports:',
|
|
' - ' . self::yamlQuote($hostPort . ':3306'),
|
|
' healthcheck:',
|
|
' test:',
|
|
' - "CMD-SHELL"',
|
|
' - "mariadb-admin ping -h 127.0.0.1 -uroot -p$${MARIADB_ROOT_PASSWORD} --silent"',
|
|
' interval: 10s',
|
|
' timeout: 5s',
|
|
' retries: 12',
|
|
]);
|
|
|
|
if ($role === 'replica') {
|
|
$seedServiceName = self::composeIdentifier($serviceName . '-seed', 'mariadb-replica-seed');
|
|
$seedScript = [
|
|
'marker="/var/lib/mysql/.truckwash-replica-seeded"',
|
|
'if [ -f "$marker" ]; then',
|
|
' echo "Replica already seeded."',
|
|
' exit 0',
|
|
'fi',
|
|
'echo "Waiting for local replica..."',
|
|
'until mariadb-admin ping -h ' . self::shellArg($serviceName) . ' -uroot -p"$MARIADB_ROOT_PASSWORD" --silent; do sleep 2; done',
|
|
'echo "Importing seed from primary..."',
|
|
'mariadb-dump --host="$MARIADB_PRIMARY_HOST" --port="$MARIADB_PRIMARY_PORT" --user="$MARIADB_PRIMARY_ADMIN_USER" --password="$MARIADB_PRIMARY_ADMIN_PASSWORD" --single-transaction --quick --routines --triggers --events --gtid --master-data=2 ' . self::mariaDbSchemaOnlyDumpIgnoreArgs('$MARIADB_SEED_DATABASE') . ' --databases "$MARIADB_SEED_DATABASE" | mariadb --host=' . self::shellArg($serviceName) . ' --user=root --password="$MARIADB_ROOT_PASSWORD"',
|
|
'for table in ' . implode(' ', self::MARIADB_SCHEMA_ONLY_TABLES) . '; do',
|
|
' mariadb-dump --host="$MARIADB_PRIMARY_HOST" --port="$MARIADB_PRIMARY_PORT" --user="$MARIADB_PRIMARY_ADMIN_USER" --password="$MARIADB_PRIMARY_ADMIN_PASSWORD" --single-transaction --quick --no-data "$MARIADB_SEED_DATABASE" "$table" | mariadb --host=' . self::shellArg($serviceName) . ' --user=root --password="$MARIADB_ROOT_PASSWORD" "$MARIADB_SEED_DATABASE" || true',
|
|
'done',
|
|
'touch "$marker"',
|
|
'echo "Replica seed completed."',
|
|
];
|
|
|
|
$lines = array_merge($lines, [
|
|
' ' . $seedServiceName . ':',
|
|
' image: ' . self::yamlQuote($image),
|
|
' restart: "no"',
|
|
' depends_on:',
|
|
' ' . $serviceName . ':',
|
|
' condition: service_healthy',
|
|
' environment:',
|
|
' MARIADB_ROOT_PASSWORD: "${MARIADB_ROOT_PASSWORD:?set MARIADB_ROOT_PASSWORD}"',
|
|
' MARIADB_PRIMARY_HOST: "${MARIADB_PRIMARY_HOST:?set MARIADB_PRIMARY_HOST}"',
|
|
' MARIADB_PRIMARY_PORT: "${MARIADB_PRIMARY_PORT:-3306}"',
|
|
' MARIADB_PRIMARY_ADMIN_USER: "${MARIADB_PRIMARY_ADMIN_USER:-root}"',
|
|
' MARIADB_PRIMARY_ADMIN_PASSWORD: "${MARIADB_PRIMARY_ADMIN_PASSWORD:?set MARIADB_PRIMARY_ADMIN_PASSWORD}"',
|
|
' MARIADB_SEED_DATABASE: ' . self::yamlQuote($database),
|
|
' volumes:',
|
|
' - ' . $volumeName . ':/var/lib/mysql',
|
|
' entrypoint:',
|
|
' - /bin/sh',
|
|
' - -ec',
|
|
' - |',
|
|
]);
|
|
foreach ($seedScript as $scriptLine) {
|
|
$lines[] = ' ' . $scriptLine;
|
|
}
|
|
}
|
|
|
|
$lines = array_merge($lines, [
|
|
'volumes:',
|
|
' ' . $volumeName . ':',
|
|
]);
|
|
|
|
$steps = [
|
|
'Deploy this compose file as a normal Docker Compose or Coolify compose service.',
|
|
'Keep server-id unique across the MariaDB primary and every replica.',
|
|
'Create or store a replication user on the primary with REPLICATION SLAVE privileges.',
|
|
];
|
|
if ($role === 'replica') {
|
|
$steps[] = 'Fill MARIADB_PRIMARY_ADMIN_PASSWORD in the generated .env file.';
|
|
$steps[] = 'Deploy the compose file and wait for the seed service to complete successfully.';
|
|
$steps[] = 'Test the connection, then save and provision the replica.';
|
|
} else {
|
|
$steps[] = 'Add the primary credentials in the superuser UI after the service is reachable.';
|
|
}
|
|
|
|
$seedCommand = implode(' ', [
|
|
'mariadb-dump',
|
|
'--host=' . self::shellArg($primaryHost),
|
|
'--port=' . $primaryPort,
|
|
'--user=<primary-admin-user>',
|
|
'--password',
|
|
'--single-transaction',
|
|
'--quick',
|
|
'--routines',
|
|
'--triggers',
|
|
'--events',
|
|
'--gtid',
|
|
'--master-data=2',
|
|
...self::mariaDbSchemaOnlySeedCommandIgnoreArgs($database),
|
|
'--databases',
|
|
self::shellArg($database),
|
|
'|',
|
|
'mariadb',
|
|
'--host=<replica-host>',
|
|
'--port=' . $hostPort,
|
|
'--user=root',
|
|
'--password',
|
|
]);
|
|
$envLines = [
|
|
'MARIADB_ROOT_PASSWORD=' . $rootPassword,
|
|
'MARIADB_PASSWORD=' . $applicationPassword,
|
|
];
|
|
if ($role === 'replica') {
|
|
$envLines[] = 'MARIADB_PRIMARY_HOST=' . $primaryHost;
|
|
$envLines[] = 'MARIADB_PRIMARY_PORT=' . $primaryPort;
|
|
$envLines[] = 'MARIADB_PRIMARY_ADMIN_USER=root';
|
|
$envLines[] = 'MARIADB_PRIMARY_ADMIN_PASSWORD=';
|
|
}
|
|
|
|
return [
|
|
'kind' => self::KIND_DATABASE,
|
|
'engine' => 'mariadb',
|
|
'role' => $role,
|
|
'service_name' => $serviceName,
|
|
'host_port' => $hostPort,
|
|
'server_id' => $serverId,
|
|
'compose' => implode("\n", $lines) . "\n",
|
|
'env' => implode("\n", $envLines) . "\n",
|
|
'seed_command' => $role === 'replica' ? $seedCommand : '',
|
|
'credentials' => [
|
|
'label' => $serviceName,
|
|
'host' => '',
|
|
'port' => $hostPort,
|
|
'database' => $database,
|
|
'username' => $username,
|
|
'password' => $applicationPassword,
|
|
'admin_username' => 'root',
|
|
'admin_password' => $rootPassword,
|
|
'replication_username' => $replicationUsername,
|
|
'replication_password' => $replicationPassword,
|
|
'ssl_mode' => 'DISABLED',
|
|
'allow_preseeded_replica' => $role === 'replica',
|
|
],
|
|
'steps' => $steps,
|
|
];
|
|
}
|
|
|
|
private static function redisComposeTemplate(array $input, string $role): array
|
|
{
|
|
$serviceName = self::composeIdentifier($input['service_name'] ?? null, 'redis-' . $role);
|
|
$volumeName = self::composeIdentifier($input['volume_name'] ?? null, $serviceName . '-data');
|
|
$image = self::composeImage($input['image'] ?? null, 'redis:7');
|
|
$hostPort = self::boundedInt($input['host_port'] ?? null, $role === 'primary' ? 6379 : 6380, 1, 65535);
|
|
$primaryHost = self::composeScalar($input['primary_host'] ?? null, 'redis-primary');
|
|
$primaryPort = self::boundedInt($input['primary_port'] ?? null, 6379, 1, 65535);
|
|
$redisPassword = self::composePassword($input['password'] ?? null);
|
|
$primaryPassword = self::composePassword($input['primary_password'] ?? null);
|
|
|
|
$command = [
|
|
'redis-server',
|
|
'--appendonly',
|
|
'yes',
|
|
'--requirepass',
|
|
'${REDIS_PASSWORD:?set REDIS_PASSWORD}',
|
|
];
|
|
if ($role === 'replica') {
|
|
array_push(
|
|
$command,
|
|
'--replicaof',
|
|
$primaryHost,
|
|
(string)$primaryPort,
|
|
'--masterauth',
|
|
'${REDIS_PRIMARY_PASSWORD:?set REDIS_PRIMARY_PASSWORD}'
|
|
);
|
|
}
|
|
|
|
$lines = [
|
|
'services:',
|
|
' ' . $serviceName . ':',
|
|
' image: ' . self::yamlQuote($image),
|
|
' restart: unless-stopped',
|
|
' environment:',
|
|
' REDIS_PASSWORD: "${REDIS_PASSWORD:?set REDIS_PASSWORD}"',
|
|
];
|
|
if ($role === 'replica') {
|
|
$lines[] = ' REDIS_PRIMARY_PASSWORD: "${REDIS_PRIMARY_PASSWORD:?set REDIS_PRIMARY_PASSWORD}"';
|
|
}
|
|
$lines[] = ' command:';
|
|
|
|
foreach ($command as $argument) {
|
|
$lines[] = ' - ' . self::yamlQuote($argument);
|
|
}
|
|
|
|
$lines = array_merge($lines, [
|
|
' volumes:',
|
|
' - ' . $volumeName . ':/data',
|
|
' ports:',
|
|
' - ' . self::yamlQuote($hostPort . ':6379'),
|
|
' healthcheck:',
|
|
' test:',
|
|
' - "CMD-SHELL"',
|
|
' - "redis-cli -a $${REDIS_PASSWORD} ping | grep PONG"',
|
|
' interval: 10s',
|
|
' timeout: 5s',
|
|
' retries: 12',
|
|
'volumes:',
|
|
' ' . $volumeName . ':',
|
|
]);
|
|
|
|
$steps = [
|
|
'Deploy this compose file as a normal Docker Compose or Coolify compose service.',
|
|
'Add the Redis credentials in the superuser UI after the service is reachable.',
|
|
];
|
|
if ($role === 'replica') {
|
|
$steps[] = 'Use the current Redis primary host and password for REDIS_PRIMARY_PASSWORD, then run Test in the superuser UI.';
|
|
}
|
|
|
|
return [
|
|
'kind' => self::KIND_REDIS,
|
|
'engine' => 'redis',
|
|
'role' => $role,
|
|
'service_name' => $serviceName,
|
|
'host_port' => $hostPort,
|
|
'compose' => implode("\n", $lines) . "\n",
|
|
'env' => implode("\n", [
|
|
'REDIS_PASSWORD=' . $redisPassword,
|
|
...($role === 'replica' ? ['REDIS_PRIMARY_PASSWORD=' . $primaryPassword] : []),
|
|
]) . "\n",
|
|
'credentials' => [
|
|
'label' => $serviceName,
|
|
'host' => '',
|
|
'port' => $hostPort,
|
|
'database' => 0,
|
|
'username' => '',
|
|
'password' => $redisPassword,
|
|
],
|
|
'steps' => $steps,
|
|
];
|
|
}
|
|
|
|
private static function normalizeComposeRole(string $role): string
|
|
{
|
|
$role = strtolower(trim($role));
|
|
if (in_array($role, ['primary', 'replica'], true)) {
|
|
return $role;
|
|
}
|
|
throw new RuntimeException('Unsupported compose role.');
|
|
}
|
|
|
|
private static function composeIdentifier(mixed $value, string $fallback): string
|
|
{
|
|
$identifier = strtolower(trim((string)$value));
|
|
$identifier = (string)preg_replace('/[^a-z0-9_.-]+/', '-', $identifier);
|
|
$identifier = trim($identifier, '-_.');
|
|
return $identifier !== '' ? $identifier : $fallback;
|
|
}
|
|
|
|
private static function composeScalar(mixed $value, string $fallback): string
|
|
{
|
|
$scalar = trim((string)$value);
|
|
return $scalar !== '' ? $scalar : $fallback;
|
|
}
|
|
|
|
private static function composePassword(mixed $value): string
|
|
{
|
|
$password = trim((string)$value);
|
|
if ($password !== '') {
|
|
return $password;
|
|
}
|
|
|
|
return self::generateSecret(24);
|
|
}
|
|
|
|
private static function generateSecret(int $bytes): string
|
|
{
|
|
return rtrim(strtr(base64_encode(random_bytes($bytes)), '+/', '-_'), '=');
|
|
}
|
|
|
|
private static function composeImage(mixed $value, string $fallback): string
|
|
{
|
|
$image = trim((string)$value);
|
|
if ($image === '' || preg_match('/^[a-zA-Z0-9._:\/-]+$/', $image) !== 1) {
|
|
return $fallback;
|
|
}
|
|
return $image;
|
|
}
|
|
|
|
private static function boundedInt(mixed $value, int $fallback, int $min, int $max): int
|
|
{
|
|
if (filter_var($value, FILTER_VALIDATE_INT) === false) {
|
|
return $fallback;
|
|
}
|
|
return max($min, min($max, (int)$value));
|
|
}
|
|
|
|
private static function yamlQuote(mixed $value): string
|
|
{
|
|
$encoded = json_encode((string)$value, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
|
|
return is_string($encoded) ? $encoded : '""';
|
|
}
|
|
|
|
private static function shellArg(string $value): string
|
|
{
|
|
return "'" . str_replace("'", "'\"'\"'", $value) . "'";
|
|
}
|
|
|
|
private static function mariaDbSchemaOnlyDumpIgnoreArgs(string $databaseExpression): string
|
|
{
|
|
return implode(' ', array_map(
|
|
static fn(string $tableName): string => '--ignore-table="' . $databaseExpression . '.' . $tableName . '"',
|
|
self::MARIADB_SCHEMA_ONLY_TABLES
|
|
));
|
|
}
|
|
|
|
private static function mariaDbSchemaOnlySeedCommandIgnoreArgs(string $database): array
|
|
{
|
|
return array_map(
|
|
static fn(string $tableName): string => '--ignore-table=' . self::shellArg($database . '.' . $tableName),
|
|
self::MARIADB_SCHEMA_ONLY_TABLES
|
|
);
|
|
}
|
|
|
|
private static function quoteIdentifier(string $identifier): string
|
|
{
|
|
return '`' . str_replace('`', '``', $identifier) . '`';
|
|
}
|
|
|
|
private static function sqlString(mysqli $connection, string $value): string
|
|
{
|
|
return "'" . $connection->real_escape_string($value) . "'";
|
|
}
|
|
|
|
public static function mysqlGtidIntervalCount(string $gtidSet): int
|
|
{
|
|
$count = 0;
|
|
foreach (self::parseMysqlGtidSet($gtidSet) as $intervals) {
|
|
foreach ($intervals as [$start, $end]) {
|
|
$count += max(0, $end - $start + 1);
|
|
}
|
|
}
|
|
return $count;
|
|
}
|
|
|
|
public static function mysqlGtidCoveragePercent(string $sourceSet, string $executedSet): float
|
|
{
|
|
$source = self::parseMysqlGtidSet($sourceSet);
|
|
$executed = self::parseMysqlGtidSet($executedSet);
|
|
$total = 0;
|
|
$covered = 0;
|
|
|
|
foreach ($source as $uuid => $sourceIntervals) {
|
|
foreach ($sourceIntervals as [$sourceStart, $sourceEnd]) {
|
|
$total += max(0, $sourceEnd - $sourceStart + 1);
|
|
foreach ($executed[$uuid] ?? [] as [$executedStart, $executedEnd]) {
|
|
$start = max($sourceStart, $executedStart);
|
|
$end = min($sourceEnd, $executedEnd);
|
|
if ($end >= $start) {
|
|
$covered += $end - $start + 1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if ($total === 0) {
|
|
return 100.0;
|
|
}
|
|
|
|
return round(min(100, max(0, ($covered / $total) * 100)), 2);
|
|
}
|
|
|
|
public static function redisOffsetPercent(int $primaryOffset, int $replicaOffset): float
|
|
{
|
|
if ($primaryOffset <= 0) {
|
|
return 100.0;
|
|
}
|
|
|
|
return round(min(100, max(0, ($replicaOffset / $primaryOffset) * 100)), 2);
|
|
}
|
|
|
|
public static function mariadbGtidCoveragePercent(string $sourceSet, string $replicaSet): float
|
|
{
|
|
$source = self::parseMariaDbGtidSet($sourceSet);
|
|
$replica = self::parseMariaDbGtidSet($replicaSet);
|
|
$total = array_sum($source);
|
|
if ($total <= 0) {
|
|
return 100.0;
|
|
}
|
|
|
|
$covered = 0;
|
|
foreach ($source as $domain => $sourceSequence) {
|
|
$covered += min($sourceSequence, $replica[$domain] ?? 0);
|
|
}
|
|
|
|
return round(min(100, max(0, ($covered / $total) * 100)), 2);
|
|
}
|
|
|
|
private static function parseMariaDbGtidSet(string $gtidSet): array
|
|
{
|
|
$positions = [];
|
|
foreach (explode(',', trim($gtidSet)) as $gtid) {
|
|
$gtid = trim($gtid);
|
|
if ($gtid === '') {
|
|
continue;
|
|
}
|
|
|
|
$parts = explode('-', $gtid);
|
|
if (count($parts) !== 3) {
|
|
continue;
|
|
}
|
|
|
|
[$domain, , $sequence] = array_map('intval', $parts);
|
|
if ($sequence <= 0) {
|
|
continue;
|
|
}
|
|
|
|
$positions[$domain] = max($positions[$domain] ?? 0, $sequence);
|
|
}
|
|
|
|
return $positions;
|
|
}
|
|
|
|
private static function parseMysqlGtidSet(string $gtidSet): array
|
|
{
|
|
$parsed = [];
|
|
foreach (explode(',', trim($gtidSet)) as $uuidSet) {
|
|
$uuidSet = trim($uuidSet);
|
|
if ($uuidSet === '') {
|
|
continue;
|
|
}
|
|
|
|
$parts = explode(':', $uuidSet);
|
|
if (count($parts) < 2) {
|
|
continue;
|
|
}
|
|
|
|
$uuid = strtolower(array_shift($parts));
|
|
foreach ($parts as $interval) {
|
|
if (str_contains($interval, '-')) {
|
|
[$start, $end] = array_map('intval', explode('-', $interval, 2));
|
|
} else {
|
|
$start = $end = (int)$interval;
|
|
}
|
|
if ($start <= 0 || $end <= 0) {
|
|
continue;
|
|
}
|
|
if ($end < $start) {
|
|
[$start, $end] = [$end, $start];
|
|
}
|
|
$parsed[$uuid][] = [$start, $end];
|
|
}
|
|
}
|
|
|
|
foreach ($parsed as $uuid => $intervals) {
|
|
usort($intervals, static fn(array $a, array $b): int => $a[0] <=> $b[0]);
|
|
$merged = [];
|
|
foreach ($intervals as [$start, $end]) {
|
|
$lastIndex = count($merged) - 1;
|
|
if ($lastIndex >= 0 && $start <= $merged[$lastIndex][1] + 1) {
|
|
$merged[$lastIndex][1] = max($merged[$lastIndex][1], $end);
|
|
continue;
|
|
}
|
|
$merged[] = [$start, $end];
|
|
}
|
|
$parsed[$uuid] = $merged;
|
|
}
|
|
|
|
return $parsed;
|
|
}
|
|
|
|
private function provisionDatabaseHost(array $host, int $operationId): array
|
|
{
|
|
$primary = $this->primaryHost(self::KIND_DATABASE);
|
|
if ($primary === null) {
|
|
throw new RuntimeException('No database primary is registered.');
|
|
}
|
|
|
|
$options = $this->decodeOptions($host);
|
|
$usePreseededReplica = !empty($options['allow_preseeded_replica']);
|
|
$targetStatus = $this->testDatabaseHost(array_merge($host, [
|
|
'test_connectivity_only' => true,
|
|
'connect_without_database' => $usePreseededReplica,
|
|
]));
|
|
$primaryStatus = $this->testDatabaseHost($primary);
|
|
$blockers = array_merge($targetStatus['blockers'], $primaryStatus['blockers']);
|
|
$targetEngine = self::databaseEngine($targetStatus['raw'] ?? []);
|
|
$primaryEngine = self::databaseEngine($primaryStatus['raw'] ?? []);
|
|
|
|
if (($targetStatus['raw']['server_id'] ?? null) !== null
|
|
&& ($primaryStatus['raw']['server_id'] ?? null) !== null
|
|
&& (int)$targetStatus['raw']['server_id'] === (int)$primaryStatus['raw']['server_id']) {
|
|
$blockers[] = 'Database replica must have a unique server_id.';
|
|
}
|
|
if ($targetEngine !== $primaryEngine) {
|
|
$blockers[] = 'Database primary and replica must use the same engine family.';
|
|
}
|
|
|
|
$cloneReady = (bool)($targetStatus['raw']['clone_plugin_active'] ?? false);
|
|
$primaryCloneReady = (bool)($primaryStatus['raw']['clone_plugin_active'] ?? false);
|
|
if ($targetEngine === 'mariadb' && !$usePreseededReplica) {
|
|
$blockers[] = 'MariaDB replicas must be safely seeded before managed replication can be configured.';
|
|
}
|
|
if ($targetEngine === 'mysql' && !$usePreseededReplica) {
|
|
if (!$cloneReady) {
|
|
$blockers[] = 'MySQL Clone plugin is not active on the target. Set allow_preseeded_replica only after the target has been safely seeded.';
|
|
}
|
|
if (!$primaryCloneReady) {
|
|
$blockers[] = 'MySQL Clone plugin is not active on the primary donor.';
|
|
}
|
|
}
|
|
|
|
if ($blockers !== []) {
|
|
$this->storeStatus($host, array_replace($targetStatus, ['blockers' => array_values(array_unique($blockers))]));
|
|
return [
|
|
'ok' => false,
|
|
'message' => 'Database replica provisioning is blocked.',
|
|
'blockers' => array_values(array_unique($blockers)),
|
|
'replication_percent' => $targetStatus['replication_percent'],
|
|
'host' => $this->publicHost($host),
|
|
];
|
|
}
|
|
|
|
$target = $this->databaseConnection($host, true, $usePreseededReplica);
|
|
try {
|
|
$primaryCredentials = $this->credentials($primary);
|
|
$hostCredentials = $this->credentials($host);
|
|
$replicationUser = $hostCredentials['replication_username']
|
|
?: ($primaryCredentials['replication_username'] ?: $primaryCredentials['username']);
|
|
$replicationPassword = $hostCredentials['replication_password']
|
|
?: ($primaryCredentials['replication_password'] ?: $primaryCredentials['password']);
|
|
$shouldManageReplicationUser = ($hostCredentials['replication_username'] ?? '') !== ''
|
|
&& ($hostCredentials['replication_password'] ?? '') !== '';
|
|
|
|
if ($usePreseededReplica && $targetEngine === 'mariadb') {
|
|
$seedContext = $this->operationContext($operationId);
|
|
$seedInProgress = ($seedContext['phase'] ?? '') !== '' && ($seedContext['phase'] ?? '') !== 'complete';
|
|
$seedBlockers = $this->databaseReplicaSeedBlockers($primary, $host, $target);
|
|
if ($seedInProgress || $seedBlockers !== []) {
|
|
$seedResult = $this->advanceMariaDbReplicaSeed($operationId, $primary, $host, $target);
|
|
$targetStatus = $seedResult['status'];
|
|
|
|
if (($seedResult['running'] ?? false) === true) {
|
|
$this->storeStatus($host, $targetStatus);
|
|
return [
|
|
'ok' => true,
|
|
'message' => $seedResult['message'],
|
|
'blockers' => [],
|
|
'replication_percent' => $targetStatus['replication_percent'],
|
|
'operation' => [
|
|
'id' => $operationId,
|
|
'status' => 'running',
|
|
'progress_percent' => $targetStatus['replication_percent'],
|
|
'message' => $seedResult['message'],
|
|
],
|
|
'host' => $this->publicHost($host),
|
|
];
|
|
}
|
|
}
|
|
} elseif ($usePreseededReplica) {
|
|
$seedBlockers = $this->databaseReplicaSeedBlockers($primary, $host, $target);
|
|
if ($seedBlockers !== []) {
|
|
$combinedBlockers = array_values(array_unique(array_merge($targetStatus['blockers'], $seedBlockers)));
|
|
$this->storeStatus($host, array_replace($targetStatus, [
|
|
'status' => 'degraded',
|
|
'replication_percent' => min(99.99, (float)$targetStatus['replication_percent']),
|
|
'blockers' => $combinedBlockers,
|
|
]));
|
|
return [
|
|
'ok' => false,
|
|
'message' => 'Database replica provisioning is blocked.',
|
|
'blockers' => $combinedBlockers,
|
|
'replication_percent' => min(99.99, (float)$targetStatus['replication_percent']),
|
|
'host' => $this->publicHost($host),
|
|
];
|
|
}
|
|
}
|
|
|
|
if ($shouldManageReplicationUser) {
|
|
$this->ensureDatabaseReplicationUser($primary, $replicationUser, $replicationPassword);
|
|
}
|
|
|
|
if ($targetEngine === 'mysql' && !$usePreseededReplica) {
|
|
$this->runMysqlClone($target, $primary, $replicationUser, $replicationPassword);
|
|
$target->close();
|
|
$target = $this->waitForDatabaseConnection($host, true, 120);
|
|
}
|
|
|
|
if ($targetEngine === 'mariadb') {
|
|
$this->configureMariaDbReplication($target, $primary, $host, $replicationUser, $replicationPassword);
|
|
} else {
|
|
$this->configureMySqlReplication($target, $primary, $replicationUser, $replicationPassword);
|
|
}
|
|
} finally {
|
|
$target->close();
|
|
}
|
|
|
|
$this->execute(
|
|
"UPDATE replication_hosts SET replication_source_id = ?, status = 'provisioning' WHERE id = ?",
|
|
'ii',
|
|
[(int)$primary['id'], (int)$host['id']]
|
|
);
|
|
|
|
$status = $this->testDatabaseHost($this->getHost(self::KIND_DATABASE, (int)$host['id']));
|
|
$this->storeStatus($host, $status);
|
|
|
|
return [
|
|
'ok' => true,
|
|
'healthy' => $status['blockers'] === [],
|
|
'message' => $targetEngine === 'mariadb'
|
|
? 'MariaDB replication was configured with GTID slave_pos.'
|
|
: 'Database replication was configured with GTID auto-positioning.',
|
|
'blockers' => $status['blockers'],
|
|
'replication_percent' => $status['replication_percent'],
|
|
'host' => $this->publicHost($this->getHost(self::KIND_DATABASE, (int)$host['id'])),
|
|
];
|
|
}
|
|
|
|
private function ensureDatabaseReplicationUser(array $primary, string $replicationUser, string $replicationPassword): void
|
|
{
|
|
if (trim($replicationUser) === '' || trim($replicationPassword) === '') {
|
|
throw new RuntimeException('Replication username and password are required.');
|
|
}
|
|
|
|
$connection = $this->databaseConnection($primary, true);
|
|
try {
|
|
$account = sprintf(
|
|
"'%s'@'%%'",
|
|
$connection->real_escape_string($replicationUser)
|
|
);
|
|
$password = $connection->real_escape_string($replicationPassword);
|
|
|
|
$this->mysqliExec($connection, "CREATE USER IF NOT EXISTS " . $account . " IDENTIFIED BY '" . $password . "'");
|
|
$this->mysqliExec($connection, "ALTER USER " . $account . " IDENTIFIED BY '" . $password . "'");
|
|
$this->mysqliExec($connection, "GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO " . $account);
|
|
$this->mysqliExec($connection, 'FLUSH PRIVILEGES');
|
|
} catch (Throwable $throwable) {
|
|
throw new RuntimeException(
|
|
'Could not create or update the replication user on the primary database. Add primary admin credentials or create the replication user manually: ' . $throwable->getMessage(),
|
|
0,
|
|
$throwable
|
|
);
|
|
} finally {
|
|
$connection->close();
|
|
}
|
|
}
|
|
|
|
private function configureMySqlReplication(mysqli $target, array $primary, string $replicationUser, string $replicationPassword): void
|
|
{
|
|
try {
|
|
$this->mysqliExec($target, 'STOP REPLICA');
|
|
} catch (Throwable) {
|
|
}
|
|
$sql = sprintf(
|
|
"CHANGE REPLICATION SOURCE TO SOURCE_HOST = '%s', SOURCE_PORT = %d, SOURCE_USER = '%s', SOURCE_PASSWORD = '%s', SOURCE_AUTO_POSITION = 1",
|
|
$target->real_escape_string((string)$primary['host']),
|
|
(int)$primary['port'],
|
|
$target->real_escape_string($replicationUser),
|
|
$target->real_escape_string($replicationPassword)
|
|
);
|
|
$this->mysqliExec($target, $sql);
|
|
$this->mysqliExec($target, 'START REPLICA');
|
|
}
|
|
|
|
private function configureMariaDbReplication(mysqli $target, array $primary, array $host, string $replicationUser, string $replicationPassword): void
|
|
{
|
|
foreach (['STOP SLAVE', 'RESET SLAVE ALL'] as $statement) {
|
|
try {
|
|
$this->mysqliExec($target, $statement);
|
|
} catch (Throwable) {
|
|
}
|
|
}
|
|
$this->configureMariaDbReplicationFilters($target, $primary, $host);
|
|
$sql = sprintf(
|
|
"CHANGE MASTER TO MASTER_HOST = '%s', MASTER_PORT = %d, MASTER_USER = '%s', MASTER_PASSWORD = '%s', MASTER_USE_GTID = slave_pos",
|
|
$target->real_escape_string((string)$primary['host']),
|
|
(int)$primary['port'],
|
|
$target->real_escape_string($replicationUser),
|
|
$target->real_escape_string($replicationPassword)
|
|
);
|
|
$this->mysqliExec($target, $sql);
|
|
$this->mysqliExec($target, 'START SLAVE');
|
|
}
|
|
|
|
private function configureMariaDbReplicationFilters(mysqli $target, array $primary, array $host): void
|
|
{
|
|
$existing = $this->mysqliSelectOne($target, "SHOW GLOBAL VARIABLES LIKE 'replicate_ignore_table'");
|
|
$filters = array_values(array_filter(array_map(
|
|
static fn(string $filter): string => trim($filter),
|
|
explode(',', (string)($existing['Value'] ?? ''))
|
|
)));
|
|
|
|
foreach ([(string)($primary['database_name'] ?? ''), (string)($host['database_name'] ?? '')] as $database) {
|
|
$database = trim($database);
|
|
if ($database !== '') {
|
|
foreach (self::MARIADB_SCHEMA_ONLY_TABLES as $tableName) {
|
|
$filters[] = $database . '.' . $tableName;
|
|
}
|
|
}
|
|
}
|
|
|
|
$filters = array_values(array_unique($filters));
|
|
if ($filters === []) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
$this->mysqliExec($target, 'SET GLOBAL replicate_ignore_table = ' . self::sqlString($target, implode(',', $filters)));
|
|
} catch (Throwable $throwable) {
|
|
throw new RuntimeException(
|
|
'Could not configure MariaDB replica schema-only table filters: ' . $throwable->getMessage(),
|
|
0,
|
|
$throwable
|
|
);
|
|
}
|
|
}
|
|
|
|
private function runMysqlClone(mysqli $target, array $primary, string $cloneUser, string $clonePassword): void
|
|
{
|
|
$donor = $target->real_escape_string((string)$primary['host'] . ':' . (int)$primary['port']);
|
|
$this->mysqliExec($target, "SET GLOBAL clone_valid_donor_list = '" . $donor . "'");
|
|
|
|
$sql = sprintf(
|
|
"CLONE INSTANCE FROM '%s'@'%s':%d IDENTIFIED BY '%s'",
|
|
$target->real_escape_string($cloneUser),
|
|
$target->real_escape_string((string)$primary['host']),
|
|
(int)$primary['port'],
|
|
$target->real_escape_string($clonePassword)
|
|
);
|
|
|
|
try {
|
|
$this->mysqliExec($target, $sql);
|
|
} catch (Throwable $throwable) {
|
|
$message = strtolower($throwable->getMessage());
|
|
if (!str_contains($message, 'lost connection') && !str_contains($message, 'server has gone away')) {
|
|
throw $throwable;
|
|
}
|
|
}
|
|
}
|
|
|
|
private function waitForDatabaseConnection(array $host, bool $admin, int $timeoutSeconds): mysqli
|
|
{
|
|
$deadline = time() + max(1, $timeoutSeconds);
|
|
$lastError = null;
|
|
|
|
do {
|
|
try {
|
|
return $this->databaseConnection($host, $admin);
|
|
} catch (Throwable $throwable) {
|
|
$lastError = $throwable;
|
|
sleep(2);
|
|
}
|
|
} while (time() < $deadline);
|
|
|
|
throw new RuntimeException('Database target did not reconnect after MySQL Clone: ' . ($lastError?->getMessage() ?? 'timeout'));
|
|
}
|
|
|
|
private function provisionRedisHost(array $host): array
|
|
{
|
|
$primary = $this->primaryHost(self::KIND_REDIS);
|
|
if ($primary === null) {
|
|
throw new RuntimeException('No Redis primary is registered.');
|
|
}
|
|
|
|
$targetStatus = $this->testRedisHost($host);
|
|
$primaryStatus = $this->testRedisHost($primary);
|
|
$blockers = array_merge($targetStatus['blockers'], $primaryStatus['blockers']);
|
|
if ($blockers !== []) {
|
|
$this->storeStatus($host, array_replace($targetStatus, ['blockers' => array_values(array_unique($blockers))]));
|
|
return [
|
|
'ok' => false,
|
|
'message' => 'Redis replica provisioning is blocked.',
|
|
'blockers' => array_values(array_unique($blockers)),
|
|
'replication_percent' => $targetStatus['replication_percent'],
|
|
'host' => $this->publicHost($host),
|
|
];
|
|
}
|
|
|
|
$client = $this->redisClient($host);
|
|
$primaryCredentials = $this->credentials($primary);
|
|
if (($primaryCredentials['username'] ?? '') !== '' && ($primaryCredentials['username'] ?? '') !== 'default') {
|
|
$client->executeRaw(['CONFIG', 'SET', 'masteruser', (string)$primaryCredentials['username']]);
|
|
}
|
|
if (($primaryCredentials['password'] ?? '') !== '') {
|
|
$client->executeRaw(['CONFIG', 'SET', 'masterauth', (string)$primaryCredentials['password']]);
|
|
}
|
|
$client->executeRaw(['REPLICAOF', (string)$primary['host'], (string)$primary['port']]);
|
|
$client->executeRaw(['CONFIG', 'REWRITE']);
|
|
|
|
$this->execute(
|
|
"UPDATE replication_hosts SET replication_source_id = ?, status = 'provisioning' WHERE id = ?",
|
|
'ii',
|
|
[(int)$primary['id'], (int)$host['id']]
|
|
);
|
|
|
|
$status = $this->testRedisHost($this->getHost(self::KIND_REDIS, (int)$host['id']));
|
|
$this->storeStatus($host, $status);
|
|
|
|
return [
|
|
'ok' => true,
|
|
'healthy' => $status['blockers'] === [],
|
|
'message' => 'Redis replication was configured.',
|
|
'blockers' => $status['blockers'],
|
|
'replication_percent' => $status['replication_percent'],
|
|
'host' => $this->publicHost($this->getHost(self::KIND_REDIS, (int)$host['id'])),
|
|
];
|
|
}
|
|
|
|
private function promoteDatabaseHost(array $host): array
|
|
{
|
|
$status = $this->testDatabaseHost($host);
|
|
if ($status['blockers'] !== [] || (float)$status['replication_percent'] < 100.0) {
|
|
throw new RuntimeException('Database promotion blocked: ' . implode(' ', $status['blockers'] ?: ['Replica is not caught up.']));
|
|
}
|
|
|
|
$oldPrimary = $this->primaryHost(self::KIND_DATABASE);
|
|
if ($oldPrimary === null) {
|
|
throw new RuntimeException('No current database primary is registered.');
|
|
}
|
|
|
|
$oldPrimaryConn = null;
|
|
$targetConn = null;
|
|
$metadataSwitched = false;
|
|
|
|
try {
|
|
$oldPrimaryConn = $this->databaseConnection($oldPrimary, true);
|
|
$oldPrimaryStatus = $this->databaseServerStatus($oldPrimaryConn);
|
|
$this->setDatabaseReadOnly($oldPrimaryConn, $oldPrimaryStatus, true);
|
|
|
|
$targetConn = $this->databaseConnection($host, true);
|
|
$targetStatus = $this->databaseServerStatus($targetConn);
|
|
$this->stopDatabaseReplication($targetConn, $targetStatus);
|
|
$this->setDatabaseReadOnly($targetConn, $targetStatus, false);
|
|
|
|
$this->switchPrimary(self::KIND_DATABASE, (int)$host['id'], (int)$oldPrimary['id']);
|
|
$metadataSwitched = true;
|
|
$this->writeBootstrapSnapshot();
|
|
} catch (Throwable $throwable) {
|
|
if (!$metadataSwitched && $oldPrimaryConn instanceof mysqli) {
|
|
try {
|
|
$this->setDatabaseReadOnly($oldPrimaryConn, $oldPrimaryStatus ?? [], false);
|
|
} catch (Throwable) {
|
|
}
|
|
}
|
|
throw $throwable;
|
|
} finally {
|
|
if ($targetConn instanceof mysqli) {
|
|
$targetConn->close();
|
|
}
|
|
if ($oldPrimaryConn instanceof mysqli) {
|
|
$oldPrimaryConn->close();
|
|
}
|
|
}
|
|
|
|
return [
|
|
'ok' => true,
|
|
'message' => 'Database replica promoted to primary.',
|
|
'primary' => $this->publicHost($this->getHost(self::KIND_DATABASE, (int)$host['id'])),
|
|
'prior_primary' => $this->publicHost($this->getHost(self::KIND_DATABASE, (int)$oldPrimary['id'], true)),
|
|
'blockers' => [],
|
|
];
|
|
}
|
|
|
|
private function setDatabaseReadOnly(mysqli $connection, array $status, bool $readOnly): void
|
|
{
|
|
$value = $readOnly ? 'ON' : 'OFF';
|
|
if (array_key_exists('super_read_only', $status)) {
|
|
try {
|
|
$this->mysqliExec($connection, 'SET GLOBAL super_read_only = ' . $value);
|
|
} catch (Throwable) {
|
|
}
|
|
}
|
|
$this->mysqliExec($connection, 'SET GLOBAL read_only = ' . $value);
|
|
}
|
|
|
|
private function stopDatabaseReplication(mysqli $connection, array $status): void
|
|
{
|
|
if (self::databaseEngine($status) === 'mariadb') {
|
|
$this->mysqliExec($connection, 'STOP SLAVE');
|
|
return;
|
|
}
|
|
|
|
$this->mysqliExec($connection, 'STOP REPLICA');
|
|
}
|
|
|
|
private function promoteRedisHost(array $host): array
|
|
{
|
|
$status = $this->testRedisHost($host);
|
|
if ($status['blockers'] !== [] || (float)$status['replication_percent'] < 100.0) {
|
|
throw new RuntimeException('Redis promotion blocked: ' . implode(' ', $status['blockers'] ?: ['Replica is not caught up.']));
|
|
}
|
|
|
|
$oldPrimary = $this->primaryHost(self::KIND_REDIS);
|
|
if ($oldPrimary === null) {
|
|
throw new RuntimeException('No current Redis primary is registered.');
|
|
}
|
|
|
|
$client = $this->redisClient($host);
|
|
$client->executeRaw(['REPLICAOF', 'NO', 'ONE']);
|
|
try {
|
|
$client->executeRaw(['CONFIG', 'REWRITE']);
|
|
} catch (Throwable) {
|
|
}
|
|
|
|
$this->switchPrimary(self::KIND_REDIS, (int)$host['id'], (int)$oldPrimary['id']);
|
|
$this->writeBootstrapSnapshot();
|
|
|
|
return [
|
|
'ok' => true,
|
|
'message' => 'Redis replica promoted to primary.',
|
|
'primary' => $this->publicHost($this->getHost(self::KIND_REDIS, (int)$host['id'])),
|
|
'prior_primary' => $this->publicHost($this->getHost(self::KIND_REDIS, (int)$oldPrimary['id'], true)),
|
|
'blockers' => [],
|
|
];
|
|
}
|
|
|
|
private function testDatabaseHost(array $host): array
|
|
{
|
|
$blockers = [];
|
|
$raw = [];
|
|
$percent = (float)(($host['role'] ?? '') === 'primary' ? 100 : 0);
|
|
$lagSeconds = null;
|
|
$reachable = true;
|
|
|
|
try {
|
|
$connection = $this->databaseConnection($host, true, !empty($host['connect_without_database']));
|
|
try {
|
|
$raw = $this->databaseServerStatus($connection);
|
|
$blockers = array_merge($blockers, self::databasePrerequisiteBlockers($raw));
|
|
|
|
if (($host['role'] ?? '') !== 'primary' && empty($host['test_connectivity_only'])) {
|
|
$primary = $this->primaryHost(self::KIND_DATABASE);
|
|
if ($primary === null) {
|
|
$blockers[] = 'No database primary is registered.';
|
|
} else {
|
|
$sourceConnection = $this->databaseConnection($primary, true);
|
|
try {
|
|
$source = $this->databaseServerStatus($sourceConnection);
|
|
$replica = $this->showReplicaStatus($connection);
|
|
$sourceEngine = self::databaseEngine($source);
|
|
$replicaEngine = self::databaseEngine($raw);
|
|
$raw['source_gtid_executed'] = self::databaseGtidPosition($source);
|
|
$raw['replica_status'] = $replica;
|
|
|
|
if ($sourceEngine !== $replicaEngine) {
|
|
$blockers[] = 'Database primary and replica must use the same engine family.';
|
|
}
|
|
$percent = $sourceEngine === 'mariadb'
|
|
? self::mariadbGtidCoveragePercent(
|
|
self::databaseGtidPosition($source),
|
|
(string)($replica['Gtid_IO_Pos'] ?? $raw['gtid_slave_pos'] ?? $raw['gtid_current_pos'] ?? '')
|
|
)
|
|
: self::mysqlGtidCoveragePercent(
|
|
(string)($source['gtid_executed'] ?? ''),
|
|
(string)($replica['Executed_Gtid_Set'] ?? $raw['gtid_executed'] ?? '')
|
|
);
|
|
$lagSeconds = isset($replica['Seconds_Behind_Source'])
|
|
? (int)$replica['Seconds_Behind_Source']
|
|
: (isset($replica['Seconds_Behind_Master']) ? (int)$replica['Seconds_Behind_Master'] : null);
|
|
|
|
$ioRunning = false;
|
|
$sqlRunning = false;
|
|
if ($replica === []) {
|
|
$blockers[] = 'Database replica status is not configured.';
|
|
} else {
|
|
$ioRunning = strtoupper((string)($replica['Replica_IO_Running'] ?? $replica['Slave_IO_Running'] ?? '')) === 'YES';
|
|
$sqlRunning = strtoupper((string)($replica['Replica_SQL_Running'] ?? $replica['Slave_SQL_Running'] ?? '')) === 'YES';
|
|
if (!$ioRunning || !$sqlRunning) {
|
|
$blockers[] = 'Database replication IO and SQL threads must both be running.';
|
|
}
|
|
if (!$ioRunning) {
|
|
$blockers[] = 'Database replication IO thread is not running.';
|
|
}
|
|
if (!$sqlRunning) {
|
|
$blockers[] = 'Database replication SQL thread is not running.';
|
|
}
|
|
if ($sourceEngine === 'mariadb' && isset($replica['Using_Gtid']) && strtoupper((string)$replica['Using_Gtid']) === 'NO') {
|
|
$blockers[] = 'MariaDB replication must use GTID mode.';
|
|
}
|
|
foreach (['Last_IO_Error', 'Last_SQL_Error', 'Last_Error'] as $errorKey) {
|
|
$error = trim((string)($replica[$errorKey] ?? ''));
|
|
if ($error !== '') {
|
|
$blockers[] = $error;
|
|
}
|
|
}
|
|
}
|
|
if (($blockers !== [] || !$ioRunning || !$sqlRunning) && $percent >= 100.0) {
|
|
$percent = 99.99;
|
|
}
|
|
} finally {
|
|
$sourceConnection->close();
|
|
}
|
|
}
|
|
}
|
|
} finally {
|
|
$connection->close();
|
|
}
|
|
} catch (Throwable $throwable) {
|
|
$reachable = false;
|
|
$blockers[] = $throwable->getMessage();
|
|
}
|
|
|
|
$blockers = array_values(array_unique(array_filter($blockers)));
|
|
return [
|
|
'status' => !$reachable ? 'down' : ($blockers === [] ? 'ok' : 'degraded'),
|
|
'replication_percent' => round($percent, 2),
|
|
'lag_seconds' => $lagSeconds,
|
|
'blockers' => $blockers,
|
|
'raw' => $raw,
|
|
'checked_at' => date('c'),
|
|
];
|
|
}
|
|
|
|
private function testRedisHost(array $host): array
|
|
{
|
|
$blockers = [];
|
|
$raw = [];
|
|
$percent = (float)(($host['role'] ?? '') === 'primary' ? 100 : 0);
|
|
$reachable = true;
|
|
|
|
try {
|
|
$client = $this->redisClient($host);
|
|
$ping = (string)$client->ping();
|
|
if (stripos($ping, 'PONG') === false && stripos($ping, 'OK') === false) {
|
|
$blockers[] = 'Redis PING did not return PONG.';
|
|
}
|
|
|
|
$role = $client->executeRaw(['ROLE']);
|
|
$info = $this->redisInfo($client);
|
|
$raw = [
|
|
'role' => $role,
|
|
'replication' => $info,
|
|
];
|
|
|
|
try {
|
|
$client->executeRaw(['CONFIG', 'GET', 'appendonly']);
|
|
} catch (Throwable $throwable) {
|
|
$blockers[] = 'Redis ACL must allow CONFIG GET/SET/REWRITE for durable replication changes.';
|
|
}
|
|
|
|
if (($host['role'] ?? '') !== 'primary' && empty($host['test_connectivity_only'])) {
|
|
$primary = $this->primaryHost(self::KIND_REDIS);
|
|
if ($primary === null) {
|
|
$blockers[] = 'No Redis primary is registered.';
|
|
} else {
|
|
$primaryClient = $this->redisClient($primary);
|
|
$primaryInfo = $this->redisInfo($primaryClient);
|
|
$primaryOffset = (int)($primaryInfo['master_repl_offset'] ?? 0);
|
|
$replicaOffset = (int)($info['slave_repl_offset'] ?? $info['master_repl_offset'] ?? 0);
|
|
$percent = self::redisOffsetPercent($primaryOffset, $replicaOffset);
|
|
$raw['primary_replication'] = $primaryInfo;
|
|
|
|
if (strtolower((string)($info['role'] ?? '')) !== 'slave') {
|
|
$blockers[] = 'Redis host is not currently a replica.';
|
|
}
|
|
if (strtolower((string)($info['master_link_status'] ?? '')) !== 'up') {
|
|
$blockers[] = 'Redis replica link to primary is not up.';
|
|
}
|
|
if ($blockers !== [] && $percent >= 100.0) {
|
|
$percent = 99.99;
|
|
}
|
|
}
|
|
}
|
|
} catch (Throwable $throwable) {
|
|
$reachable = false;
|
|
$blockers[] = $throwable->getMessage();
|
|
}
|
|
|
|
$blockers = array_values(array_unique(array_filter($blockers)));
|
|
return [
|
|
'status' => !$reachable ? 'down' : ($blockers === [] ? 'ok' : 'degraded'),
|
|
'replication_percent' => round($percent, 2),
|
|
'lag_seconds' => null,
|
|
'blockers' => $blockers,
|
|
'raw' => $raw,
|
|
'checked_at' => date('c'),
|
|
];
|
|
}
|
|
|
|
public static function databasePrerequisiteBlockers(array $status): array
|
|
{
|
|
if (self::databaseEngine($status) === 'mariadb') {
|
|
return self::mariaDbPrerequisiteBlockers($status);
|
|
}
|
|
|
|
$blockers = [];
|
|
if (strtoupper((string)($status['gtid_mode'] ?? '')) !== 'ON') {
|
|
$blockers[] = isset($status['gtid_mode'])
|
|
? 'MySQL GTID mode must be ON.'
|
|
: 'MySQL GTID mode is unavailable. Managed replication requires Oracle MySQL 8.x with GTID enabled.';
|
|
}
|
|
if (!self::mysqlBooleanEnabled($status['log_bin'] ?? null)) {
|
|
$blockers[] = isset($status['log_bin'])
|
|
? 'MySQL binary logging must be enabled.'
|
|
: 'MySQL binary logging status is unavailable.';
|
|
}
|
|
if ((int)($status['server_id'] ?? 0) <= 0) {
|
|
$blockers[] = 'MySQL server_id must be configured.';
|
|
}
|
|
if (trim((string)($status['server_uuid'] ?? '')) === '') {
|
|
$blockers[] = 'MySQL server_uuid must be available. Managed replication requires Oracle MySQL 8.x.';
|
|
}
|
|
$serverVersion = (string)($status['server_version'] ?? '');
|
|
if (!str_starts_with($serverVersion, '8.') || stripos($serverVersion, 'mariadb') !== false) {
|
|
$blockers[] = $serverVersion !== ''
|
|
? 'Oracle MySQL 8.x is required for managed replication. Current server reports ' . $serverVersion . '.'
|
|
: 'Oracle MySQL 8.x is required for managed replication.';
|
|
}
|
|
|
|
return $blockers;
|
|
}
|
|
|
|
public static function missingDatabaseTables(array $sourceTables, array $replicaTables): array
|
|
{
|
|
$source = array_values(array_unique(array_filter(array_map(
|
|
static fn(mixed $table): string => trim((string)$table),
|
|
$sourceTables
|
|
))));
|
|
$replicaLookup = array_flip(array_values(array_unique(array_filter(array_map(
|
|
static fn(mixed $table): string => trim((string)$table),
|
|
$replicaTables
|
|
)))));
|
|
|
|
return array_values(array_filter(
|
|
$source,
|
|
static fn(string $table): bool => !isset($replicaLookup[$table])
|
|
));
|
|
}
|
|
|
|
private static function mariaDbPrerequisiteBlockers(array $status): array
|
|
{
|
|
$blockers = [];
|
|
if (!self::mysqlBooleanEnabled($status['log_bin'] ?? null)) {
|
|
$blockers[] = isset($status['log_bin'])
|
|
? 'MariaDB binary logging must be enabled.'
|
|
: 'MariaDB binary logging status is unavailable.';
|
|
}
|
|
if ((int)($status['server_id'] ?? 0) <= 0) {
|
|
$blockers[] = 'MariaDB server_id must be configured.';
|
|
}
|
|
if (!self::mariaDbGtidPositionAvailable($status)) {
|
|
$blockers[] = 'MariaDB GTID position must be available.';
|
|
}
|
|
|
|
$serverVersion = (string)($status['server_version'] ?? '');
|
|
if (!self::mariaDbVersionSupported($serverVersion)) {
|
|
$blockers[] = $serverVersion !== ''
|
|
? 'MariaDB 10.6 or newer is required for managed replication. Current server reports ' . $serverVersion . '.'
|
|
: 'MariaDB 10.6 or newer is required for managed replication.';
|
|
}
|
|
|
|
return $blockers;
|
|
}
|
|
|
|
private static function databaseEngine(array $status): string
|
|
{
|
|
return stripos((string)($status['server_version'] ?? ''), 'mariadb') !== false ? 'mariadb' : 'mysql';
|
|
}
|
|
|
|
private static function databaseGtidPosition(array $status): string
|
|
{
|
|
if (self::databaseEngine($status) === 'mariadb') {
|
|
return trim((string)($status['gtid_binlog_pos'] ?? $status['gtid_current_pos'] ?? $status['gtid_slave_pos'] ?? ''));
|
|
}
|
|
|
|
return trim((string)($status['gtid_executed'] ?? ''));
|
|
}
|
|
|
|
private static function mariaDbGtidPositionAvailable(array $status): bool
|
|
{
|
|
foreach (['gtid_binlog_pos', 'gtid_current_pos', 'gtid_slave_pos'] as $key) {
|
|
if (array_key_exists($key, $status) && $status[$key] !== null) {
|
|
return true;
|
|
}
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
private static function mariaDbVersionSupported(string $serverVersion): bool
|
|
{
|
|
if (!preg_match('/(\d+)\.(\d+)/', $serverVersion, $matches)) {
|
|
return false;
|
|
}
|
|
|
|
$major = (int)$matches[1];
|
|
$minor = (int)$matches[2];
|
|
return $major > 10 || ($major === 10 && $minor >= 6);
|
|
}
|
|
|
|
private static function mysqlBooleanEnabled(mixed $value): bool
|
|
{
|
|
$normalized = strtoupper(trim((string)$value));
|
|
return in_array($normalized, ['1', 'ON', 'YES', 'TRUE'], true);
|
|
}
|
|
|
|
private function databaseServerStatus(mysqli $connection): array
|
|
{
|
|
$row = $this->mysqliSelectOne($connection, "SELECT VERSION() AS server_version");
|
|
$variables = $connection->query(
|
|
"SHOW GLOBAL VARIABLES WHERE Variable_name IN (
|
|
'gtid_mode',
|
|
'log_bin',
|
|
'server_id',
|
|
'server_uuid',
|
|
'read_only',
|
|
'super_read_only',
|
|
'gtid_executed',
|
|
'gtid_binlog_pos',
|
|
'gtid_current_pos',
|
|
'gtid_slave_pos',
|
|
'gtid_strict_mode'
|
|
)"
|
|
);
|
|
if ($variables !== false) {
|
|
while ($variable = $variables->fetch_assoc()) {
|
|
$name = strtolower((string)($variable['Variable_name'] ?? ''));
|
|
if ($name !== '') {
|
|
$row[$name] = $variable['Value'] ?? null;
|
|
}
|
|
}
|
|
}
|
|
|
|
$plugin = $this->mysqliSelectOne(
|
|
$connection,
|
|
"SELECT PLUGIN_STATUS AS plugin_status FROM information_schema.PLUGINS WHERE PLUGIN_NAME = 'clone' LIMIT 1"
|
|
);
|
|
$row['clone_plugin_active'] = strtoupper((string)($plugin['plugin_status'] ?? '')) === 'ACTIVE';
|
|
|
|
return $row;
|
|
}
|
|
|
|
private function showReplicaStatus(mysqli $connection): array
|
|
{
|
|
try {
|
|
$status = $this->mysqliSelectOne($connection, 'SHOW REPLICA STATUS');
|
|
if ($status !== []) {
|
|
return $status;
|
|
}
|
|
} catch (Throwable) {
|
|
}
|
|
|
|
return $this->mysqliSelectOne($connection, 'SHOW SLAVE STATUS');
|
|
}
|
|
|
|
private function databaseConnection(array $host, bool $admin = false, bool $connectWithoutDatabase = false): mysqli
|
|
{
|
|
$credentials = $this->credentials($host);
|
|
$username = $admin && $credentials['admin_username'] !== ''
|
|
? $credentials['admin_username']
|
|
: $credentials['username'];
|
|
$password = $admin && $credentials['admin_password'] !== ''
|
|
? $credentials['admin_password']
|
|
: $credentials['password'];
|
|
|
|
mysqli_report(MYSQLI_REPORT_ERROR | MYSQLI_REPORT_STRICT);
|
|
$connection = new mysqli(
|
|
(string)$host['host'],
|
|
$username,
|
|
$password,
|
|
$connectWithoutDatabase ? '' : (string)($host['database_name'] ?? ''),
|
|
(int)$host['port']
|
|
);
|
|
$connection->set_charset('utf8mb4');
|
|
return $connection;
|
|
}
|
|
|
|
private function databaseReplicaSeedBlockers(?array $primary, array $host, mysqli $target): array
|
|
{
|
|
if ($primary === null) {
|
|
return ['No database primary is registered.'];
|
|
}
|
|
|
|
$primaryDatabase = trim((string)($primary['database_name'] ?? ''));
|
|
$targetDatabase = trim((string)($host['database_name'] ?? ''));
|
|
if ($primaryDatabase === '' || $targetDatabase === '') {
|
|
return [];
|
|
}
|
|
|
|
$primaryConnection = $this->databaseConnection($primary, true);
|
|
try {
|
|
$primaryTables = $this->databaseTableNames($primaryConnection, $primaryDatabase);
|
|
$targetTables = $this->databaseTableNames($target, $targetDatabase);
|
|
} finally {
|
|
$primaryConnection->close();
|
|
}
|
|
|
|
if ($primaryTables === []) {
|
|
return [];
|
|
}
|
|
|
|
$missingTables = self::missingDatabaseTables($primaryTables, $targetTables);
|
|
if ($missingTables === []) {
|
|
$schemaOnlyTablesWithRows = $this->databaseSchemaOnlyTablesWithRows($target, $targetDatabase);
|
|
if ($schemaOnlyTablesWithRows === []) {
|
|
return [];
|
|
}
|
|
|
|
return [self::schemaOnlyTablesContainRowsBlocker($targetDatabase, $schemaOnlyTablesWithRows)];
|
|
}
|
|
|
|
return [self::missingDatabaseTablesBlocker($targetDatabase, $missingTables)];
|
|
}
|
|
|
|
private function advanceMariaDbReplicaSeed(int $operationId, array $primary, array $host, mysqli $target): array
|
|
{
|
|
$owner = 'replication-seed:' . $operationId;
|
|
$context = $this->operationContext($operationId);
|
|
$freezeState = application_write_freeze::state();
|
|
if (($context['phase'] ?? '') !== '' && ($freezeState['owner'] ?? null) !== $owner) {
|
|
$context = [];
|
|
}
|
|
if (self::mariaDbSeedContextRequiresFilterReset($context)) {
|
|
$context = [];
|
|
}
|
|
application_write_freeze::freeze('MariaDB replica seed is copying data.', $owner, self::MARIADB_SEED_FREEZE_TTL_SECONDS);
|
|
|
|
$source = $this->databaseConnection($primary, true);
|
|
$deadline = microtime(true) + self::MARIADB_SEED_STEP_SECONDS;
|
|
|
|
try {
|
|
$this->mysqliExec($target, 'SET SESSION FOREIGN_KEY_CHECKS = 0');
|
|
$this->mysqliExec($target, 'SET SESSION UNIQUE_CHECKS = 0');
|
|
|
|
if (($context['phase'] ?? '') === '') {
|
|
$context = $this->initializeMariaDbSeedContext($source, $target, $primary, $host);
|
|
}
|
|
|
|
while (microtime(true) < $deadline && ($context['phase'] ?? '') !== 'complete') {
|
|
if (($context['phase'] ?? '') === 'schema') {
|
|
$context = $this->advanceMariaDbSeedSchema($source, $target, $context);
|
|
continue;
|
|
}
|
|
|
|
if (($context['phase'] ?? '') === 'copy') {
|
|
$context = $this->advanceMariaDbSeedRows($source, $target, $context);
|
|
continue;
|
|
}
|
|
|
|
break;
|
|
}
|
|
|
|
$progress = self::mariaDbSeedProgress($context);
|
|
$message = self::mariaDbSeedMessage($context);
|
|
$this->updateOperationProgress($operationId, $progress, $message, $context);
|
|
|
|
if (($context['phase'] ?? '') === 'complete') {
|
|
application_write_freeze::unfreeze($owner);
|
|
return [
|
|
'running' => false,
|
|
'message' => 'MariaDB replica seed completed.',
|
|
'status' => [
|
|
'status' => 'provisioning',
|
|
'replication_percent' => 90.0,
|
|
'lag_seconds' => null,
|
|
'blockers' => [],
|
|
'raw' => ['seed' => $context],
|
|
'checked_at' => date('c'),
|
|
],
|
|
];
|
|
}
|
|
|
|
return [
|
|
'running' => true,
|
|
'message' => $message,
|
|
'status' => [
|
|
'status' => 'provisioning',
|
|
'replication_percent' => $progress,
|
|
'lag_seconds' => null,
|
|
'blockers' => ['MariaDB replica seed is running.'],
|
|
'raw' => ['seed' => $context],
|
|
'checked_at' => date('c'),
|
|
],
|
|
];
|
|
} catch (Throwable $throwable) {
|
|
application_write_freeze::unfreeze($owner);
|
|
throw $throwable;
|
|
} finally {
|
|
$source->close();
|
|
}
|
|
}
|
|
|
|
private function initializeMariaDbSeedContext(mysqli $source, mysqli $target, array $primary, array $host): array
|
|
{
|
|
$primaryDatabase = trim((string)($primary['database_name'] ?? ''));
|
|
$targetDatabase = trim((string)($host['database_name'] ?? ''));
|
|
if ($primaryDatabase === '' || $targetDatabase === '') {
|
|
throw new RuntimeException('Database names are required before a MariaDB replica can be seeded.');
|
|
}
|
|
|
|
$sourceGtid = $this->mariaDbCurrentGtid($source);
|
|
$this->prepareMariaDbReplicaTarget($target, $source, $primaryDatabase, $targetDatabase);
|
|
|
|
$tables = [];
|
|
foreach ($this->databaseTableNames($source, $primaryDatabase) as $tableName) {
|
|
$skipData = self::databaseReplicaSeedSkipsTableData($tableName);
|
|
$tables[] = [
|
|
'name' => $tableName,
|
|
'rows' => $skipData ? 0 : $this->estimatedDatabaseTableRows($source, $primaryDatabase, $tableName),
|
|
'copied' => 0,
|
|
'schema_created' => false,
|
|
'skip_data' => $skipData,
|
|
'skip_reason' => $skipData ? 'excluded from managed replication' : '',
|
|
];
|
|
}
|
|
|
|
return [
|
|
'phase' => 'schema',
|
|
'source_database' => $primaryDatabase,
|
|
'target_database' => $targetDatabase,
|
|
'source_gtid' => $sourceGtid,
|
|
'schema_index' => 0,
|
|
'copy_index' => 0,
|
|
'tables' => $tables,
|
|
'started_at' => date('c'),
|
|
'updated_at' => date('c'),
|
|
];
|
|
}
|
|
|
|
private function advanceMariaDbSeedSchema(mysqli $source, mysqli $target, array $context): array
|
|
{
|
|
$tables = $context['tables'] ?? [];
|
|
$index = (int)($context['schema_index'] ?? 0);
|
|
if (!isset($tables[$index])) {
|
|
$context['phase'] = 'copy';
|
|
$context['copy_index'] = 0;
|
|
$context['updated_at'] = date('c');
|
|
return $context;
|
|
}
|
|
|
|
$table = $tables[$index];
|
|
$this->createMariaDbReplicaTable(
|
|
$source,
|
|
$target,
|
|
(string)$context['source_database'],
|
|
(string)$context['target_database'],
|
|
(string)$table['name']
|
|
);
|
|
|
|
$context['tables'][$index]['schema_created'] = true;
|
|
$context['schema_index'] = $index + 1;
|
|
$context['updated_at'] = date('c');
|
|
return $context;
|
|
}
|
|
|
|
private function advanceMariaDbSeedRows(mysqli $source, mysqli $target, array $context): array
|
|
{
|
|
$tables = $context['tables'] ?? [];
|
|
$index = (int)($context['copy_index'] ?? 0);
|
|
if (!isset($tables[$index])) {
|
|
if (trim((string)($context['source_gtid'] ?? '')) !== '') {
|
|
$this->mysqliExec($target, 'SET GLOBAL gtid_slave_pos = ' . self::sqlString($target, (string)$context['source_gtid']));
|
|
}
|
|
$this->mysqliExec($target, 'SET SESSION FOREIGN_KEY_CHECKS = 1');
|
|
$this->mysqliExec($target, 'SET SESSION UNIQUE_CHECKS = 1');
|
|
$context['phase'] = 'complete';
|
|
$context['updated_at'] = date('c');
|
|
$context['completed_at'] = date('c');
|
|
return $context;
|
|
}
|
|
|
|
$tableName = (string)($tables[$index]['name'] ?? '');
|
|
$copied = (int)($tables[$index]['copied'] ?? 0);
|
|
if (!empty($tables[$index]['skip_data'])) {
|
|
$context['tables'][$index]['copied'] = (int)($tables[$index]['rows'] ?? 0);
|
|
$context['copy_index'] = $index + 1;
|
|
$context['updated_at'] = date('c');
|
|
return $context;
|
|
}
|
|
|
|
$copiedNow = $this->copyMariaDbReplicaTableRowsChunk(
|
|
$source,
|
|
$target,
|
|
(string)$context['source_database'],
|
|
(string)$context['target_database'],
|
|
$tableName,
|
|
$copied,
|
|
self::MARIADB_SEED_BATCH_ROWS
|
|
);
|
|
|
|
$context['tables'][$index]['copied'] = $copied + $copiedNow;
|
|
if ($copiedNow < self::MARIADB_SEED_BATCH_ROWS) {
|
|
$context['copy_index'] = $index + 1;
|
|
}
|
|
$context['updated_at'] = date('c');
|
|
return $context;
|
|
}
|
|
|
|
private static function mariaDbSeedProgress(array $context): float
|
|
{
|
|
$tables = is_array($context['tables'] ?? null) ? $context['tables'] : [];
|
|
if (($context['phase'] ?? '') === 'complete') {
|
|
return 90.0;
|
|
}
|
|
if ($tables === []) {
|
|
return 5.0;
|
|
}
|
|
|
|
$schemaCount = count($tables);
|
|
$schemaDone = min($schemaCount, (int)($context['schema_index'] ?? 0));
|
|
$schemaProgress = $schemaCount > 0 ? ($schemaDone / $schemaCount) * 20.0 : 20.0;
|
|
|
|
$totalRows = 0;
|
|
$copiedRows = 0;
|
|
foreach ($tables as $table) {
|
|
if (!empty($table['skip_data'])) {
|
|
continue;
|
|
}
|
|
$rows = max(1, (int)($table['rows'] ?? 0));
|
|
$totalRows += $rows;
|
|
$copiedRows += min($rows, (int)($table['copied'] ?? 0));
|
|
}
|
|
$copyProgress = $totalRows > 0 ? ($copiedRows / $totalRows) * 65.0 : 0.0;
|
|
|
|
return round(min(89.0, 5.0 + $schemaProgress + $copyProgress), 2);
|
|
}
|
|
|
|
private static function mariaDbSeedMessage(array $context): string
|
|
{
|
|
$tables = is_array($context['tables'] ?? null) ? $context['tables'] : [];
|
|
if (($context['phase'] ?? '') === 'schema') {
|
|
return 'Creating replica schema ' . min(count($tables), (int)($context['schema_index'] ?? 0)) . ' of ' . count($tables) . '.';
|
|
}
|
|
if (($context['phase'] ?? '') === 'copy') {
|
|
$index = (int)($context['copy_index'] ?? 0);
|
|
$table = $tables[$index]['name'] ?? 'table data';
|
|
if (!empty($tables[$index]['skip_data'])) {
|
|
return 'Skipping replica data for ' . $table . '.';
|
|
}
|
|
return 'Copying replica data for ' . $table . '.';
|
|
}
|
|
if (($context['phase'] ?? '') === 'complete') {
|
|
return 'Replica seed completed.';
|
|
}
|
|
return 'Preparing replica seed.';
|
|
}
|
|
|
|
private function seedMariaDbReplicaFromPrimary(array $primary, array $host, mysqli $target): void
|
|
{
|
|
$primaryDatabase = trim((string)($primary['database_name'] ?? ''));
|
|
$targetDatabase = trim((string)($host['database_name'] ?? ''));
|
|
if ($primaryDatabase === '' || $targetDatabase === '') {
|
|
throw new RuntimeException('Database names are required before a MariaDB replica can be seeded.');
|
|
}
|
|
|
|
$source = $this->databaseConnection($primary, true);
|
|
$readLockAcquired = false;
|
|
$transactionStarted = false;
|
|
|
|
try {
|
|
$source->query('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
|
|
$source->query('FLUSH TABLES WITH READ LOCK');
|
|
$readLockAcquired = true;
|
|
$source->query('START TRANSACTION WITH CONSISTENT SNAPSHOT');
|
|
$transactionStarted = true;
|
|
$sourceGtid = $this->mariaDbCurrentGtid($source);
|
|
$source->query('UNLOCK TABLES');
|
|
$readLockAcquired = false;
|
|
|
|
$this->prepareMariaDbReplicaTarget($target, $source, $primaryDatabase, $targetDatabase);
|
|
foreach ($this->databaseTableNames($source, $primaryDatabase) as $tableName) {
|
|
$this->createMariaDbReplicaTable($source, $target, $primaryDatabase, $targetDatabase, $tableName);
|
|
$this->copyMariaDbReplicaTableRows($source, $target, $primaryDatabase, $targetDatabase, $tableName);
|
|
}
|
|
|
|
if ($sourceGtid !== '') {
|
|
$this->mysqliExec($target, 'SET GLOBAL gtid_slave_pos = ' . self::sqlString($target, $sourceGtid));
|
|
}
|
|
$this->mysqliExec($target, 'SET SESSION FOREIGN_KEY_CHECKS = 1');
|
|
|
|
$source->query('COMMIT');
|
|
$transactionStarted = false;
|
|
} catch (Throwable $throwable) {
|
|
if ($readLockAcquired) {
|
|
try {
|
|
$source->query('UNLOCK TABLES');
|
|
} catch (Throwable) {
|
|
}
|
|
}
|
|
if ($transactionStarted) {
|
|
try {
|
|
$source->query('ROLLBACK');
|
|
} catch (Throwable) {
|
|
}
|
|
}
|
|
throw new RuntimeException('MariaDB replica seed failed: ' . $throwable->getMessage(), 0, $throwable);
|
|
} finally {
|
|
$source->close();
|
|
}
|
|
}
|
|
|
|
private function mariaDbCurrentGtid(mysqli $source): string
|
|
{
|
|
foreach (['gtid_binlog_pos', 'gtid_current_pos'] as $variable) {
|
|
$row = $this->mysqliSelectOne($source, "SELECT @@GLOBAL.$variable AS value");
|
|
$value = trim((string)($row['value'] ?? ''));
|
|
if ($value !== '') {
|
|
return $value;
|
|
}
|
|
}
|
|
|
|
return '';
|
|
}
|
|
|
|
private function prepareMariaDbReplicaTarget(mysqli $target, mysqli $source, string $primaryDatabase, string $targetDatabase): void
|
|
{
|
|
foreach (['STOP SLAVE', 'RESET SLAVE ALL'] as $statement) {
|
|
try {
|
|
$this->mysqliExec($target, $statement);
|
|
} catch (Throwable) {
|
|
}
|
|
}
|
|
|
|
$this->mysqliExec($target, 'SET SESSION FOREIGN_KEY_CHECKS = 0');
|
|
$this->mysqliExec($target, 'DROP DATABASE IF EXISTS ' . self::quoteIdentifier($targetDatabase));
|
|
$this->mysqliExec($target, $this->createDatabaseSql($source, $primaryDatabase, $targetDatabase));
|
|
$this->mysqliExec($target, 'USE ' . self::quoteIdentifier($targetDatabase));
|
|
|
|
try {
|
|
$this->mysqliExec($target, 'RESET MASTER');
|
|
} catch (Throwable) {
|
|
}
|
|
try {
|
|
$this->mysqliExec($target, "SET GLOBAL gtid_slave_pos = ''");
|
|
} catch (Throwable) {
|
|
}
|
|
}
|
|
|
|
private function createDatabaseSql(mysqli $source, string $primaryDatabase, string $targetDatabase): string
|
|
{
|
|
$stmt = $source->prepare(
|
|
'SELECT DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME
|
|
FROM information_schema.SCHEMATA
|
|
WHERE SCHEMA_NAME = ?
|
|
LIMIT 1'
|
|
);
|
|
if ($stmt === false) {
|
|
throw new RuntimeException('Could not prepare database schema lookup.');
|
|
}
|
|
|
|
$stmt->bind_param('s', $primaryDatabase);
|
|
$stmt->execute();
|
|
$result = $stmt->get_result();
|
|
$row = $result ? $result->fetch_assoc() : null;
|
|
$stmt->close();
|
|
|
|
$charset = preg_replace('/[^a-zA-Z0-9_]/', '', (string)($row['DEFAULT_CHARACTER_SET_NAME'] ?? 'utf8mb4')) ?: 'utf8mb4';
|
|
$collation = preg_replace('/[^a-zA-Z0-9_]/', '', (string)($row['DEFAULT_COLLATION_NAME'] ?? 'utf8mb4_unicode_ci')) ?: 'utf8mb4_unicode_ci';
|
|
|
|
return 'CREATE DATABASE ' . self::quoteIdentifier($targetDatabase)
|
|
. ' CHARACTER SET ' . $charset
|
|
. ' COLLATE ' . $collation;
|
|
}
|
|
|
|
private function createMariaDbReplicaTable(mysqli $source, mysqli $target, string $primaryDatabase, string $targetDatabase, string $tableName): void
|
|
{
|
|
$create = $this->mysqliSelectOne(
|
|
$source,
|
|
'SHOW CREATE TABLE ' . self::quoteIdentifier($primaryDatabase) . '.' . self::quoteIdentifier($tableName)
|
|
);
|
|
$createSql = (string)($create['Create Table'] ?? '');
|
|
if ($createSql === '') {
|
|
throw new RuntimeException('Could not read CREATE TABLE for ' . $primaryDatabase . '.' . $tableName . '.');
|
|
}
|
|
|
|
$this->mysqliExec($target, 'USE ' . self::quoteIdentifier($targetDatabase));
|
|
$this->mysqliExec($target, $createSql);
|
|
}
|
|
|
|
private static function mariaDbSeedContextRequiresFilterReset(array $context): bool
|
|
{
|
|
if (($context['phase'] ?? '') === '' || ($context['phase'] ?? '') === 'complete') {
|
|
return false;
|
|
}
|
|
|
|
foreach (($context['tables'] ?? []) as $table) {
|
|
if (self::databaseReplicaSeedSkipsTableData((string)($table['name'] ?? ''))
|
|
&& empty($table['skip_data'])) {
|
|
return true;
|
|
}
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
private function copyMariaDbReplicaTableRows(mysqli $source, mysqli $target, string $primaryDatabase, string $targetDatabase, string $tableName): void
|
|
{
|
|
if (self::databaseReplicaSeedSkipsTableData($tableName)) {
|
|
return;
|
|
}
|
|
|
|
$offset = 0;
|
|
do {
|
|
$copied = $this->copyMariaDbReplicaTableRowsChunk(
|
|
$source,
|
|
$target,
|
|
$primaryDatabase,
|
|
$targetDatabase,
|
|
$tableName,
|
|
$offset,
|
|
self::MARIADB_SEED_BATCH_ROWS
|
|
);
|
|
$offset += $copied;
|
|
} while ($copied >= self::MARIADB_SEED_BATCH_ROWS);
|
|
}
|
|
|
|
private function copyMariaDbReplicaTableRowsChunk(
|
|
mysqli $source,
|
|
mysqli $target,
|
|
string $primaryDatabase,
|
|
string $targetDatabase,
|
|
string $tableName,
|
|
int $offset,
|
|
int $limit
|
|
): int {
|
|
$columnNames = $this->databaseWritableColumnNames($source, $primaryDatabase, $tableName);
|
|
if ($columnNames === []) {
|
|
return 0;
|
|
}
|
|
|
|
$quotedColumns = array_map(static fn(string $column): string => self::quoteIdentifier($column), $columnNames);
|
|
$primaryKeyColumns = $this->databasePrimaryKeyColumnNames($source, $primaryDatabase, $tableName);
|
|
$orderSql = $primaryKeyColumns !== []
|
|
? ' ORDER BY ' . implode(', ', array_map(static fn(string $column): string => self::quoteIdentifier($column), $primaryKeyColumns))
|
|
: '';
|
|
$result = $source->query(
|
|
'SELECT ' . implode(', ', $quotedColumns)
|
|
. ' FROM ' . self::quoteIdentifier($primaryDatabase) . '.' . self::quoteIdentifier($tableName)
|
|
. $orderSql
|
|
. ' LIMIT ' . max(0, $offset) . ', ' . max(1, $limit),
|
|
MYSQLI_USE_RESULT
|
|
);
|
|
if ($result === false) {
|
|
throw new RuntimeException('Could not read rows from ' . $primaryDatabase . '.' . $tableName . '.');
|
|
}
|
|
|
|
$fields = $result->fetch_fields();
|
|
$insertPrefix = 'INSERT INTO ' . self::quoteIdentifier($targetDatabase) . '.' . self::quoteIdentifier($tableName)
|
|
. ' (' . implode(', ', $quotedColumns) . ') VALUES ';
|
|
$rows = [];
|
|
$batchSize = 200;
|
|
$copied = 0;
|
|
|
|
try {
|
|
$target->begin_transaction();
|
|
while (true) {
|
|
$row = $result->fetch_assoc();
|
|
if (!is_array($row)) {
|
|
break;
|
|
}
|
|
|
|
$values = [];
|
|
foreach ($fields as $field) {
|
|
$value = $row[$field->name] ?? null;
|
|
$values[] = $value === null ? 'NULL' : self::sqlString($target, (string)$value);
|
|
}
|
|
$rows[] = '(' . implode(', ', $values) . ')';
|
|
$copied++;
|
|
|
|
if (count($rows) >= $batchSize) {
|
|
$this->mysqliExec($target, $insertPrefix . implode(', ', $rows));
|
|
$rows = [];
|
|
}
|
|
}
|
|
|
|
if ($rows !== []) {
|
|
$this->mysqliExec($target, $insertPrefix . implode(', ', $rows));
|
|
}
|
|
$target->commit();
|
|
} catch (Throwable $throwable) {
|
|
try {
|
|
$target->rollback();
|
|
} catch (Throwable) {
|
|
}
|
|
throw $throwable;
|
|
} finally {
|
|
$result->free();
|
|
}
|
|
|
|
return $copied;
|
|
}
|
|
|
|
private function estimatedDatabaseTableRows(mysqli $connection, string $database, string $tableName): int
|
|
{
|
|
$stmt = $connection->prepare(
|
|
"SELECT TABLE_ROWS
|
|
FROM information_schema.TABLES
|
|
WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND TABLE_TYPE = 'BASE TABLE'
|
|
LIMIT 1"
|
|
);
|
|
if ($stmt === false) {
|
|
return 1;
|
|
}
|
|
|
|
$stmt->bind_param('ss', $database, $tableName);
|
|
$stmt->execute();
|
|
$result = $stmt->get_result();
|
|
$row = $result ? $result->fetch_assoc() : null;
|
|
$stmt->close();
|
|
|
|
return max(1, (int)($row['TABLE_ROWS'] ?? 1));
|
|
}
|
|
|
|
private static function databaseReplicaSeedSkipsTableData(string $tableName): bool
|
|
{
|
|
return in_array(strtolower($tableName), self::MARIADB_SCHEMA_ONLY_TABLES, true);
|
|
}
|
|
|
|
private function databaseWritableColumnNames(mysqli $connection, string $database, string $tableName): array
|
|
{
|
|
$stmt = $connection->prepare(
|
|
"SELECT COLUMN_NAME
|
|
FROM information_schema.COLUMNS
|
|
WHERE TABLE_SCHEMA = ?
|
|
AND TABLE_NAME = ?
|
|
AND EXTRA NOT LIKE '%GENERATED%'
|
|
ORDER BY ORDINAL_POSITION"
|
|
);
|
|
if ($stmt === false) {
|
|
throw new RuntimeException('Could not prepare table column lookup.');
|
|
}
|
|
|
|
$stmt->bind_param('ss', $database, $tableName);
|
|
$stmt->execute();
|
|
$result = $stmt->get_result();
|
|
$rows = $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
|
|
$stmt->close();
|
|
|
|
return array_values(array_filter(array_map(
|
|
static fn(array $row): string => (string)($row['COLUMN_NAME'] ?? ''),
|
|
$rows
|
|
)));
|
|
}
|
|
|
|
private function databasePrimaryKeyColumnNames(mysqli $connection, string $database, string $tableName): array
|
|
{
|
|
$stmt = $connection->prepare(
|
|
"SELECT COLUMN_NAME
|
|
FROM information_schema.KEY_COLUMN_USAGE
|
|
WHERE TABLE_SCHEMA = ?
|
|
AND TABLE_NAME = ?
|
|
AND CONSTRAINT_NAME = 'PRIMARY'
|
|
ORDER BY ORDINAL_POSITION"
|
|
);
|
|
if ($stmt === false) {
|
|
return [];
|
|
}
|
|
|
|
$stmt->bind_param('ss', $database, $tableName);
|
|
$stmt->execute();
|
|
$result = $stmt->get_result();
|
|
$rows = $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
|
|
$stmt->close();
|
|
|
|
return array_values(array_filter(array_map(
|
|
static fn(array $row): string => (string)($row['COLUMN_NAME'] ?? ''),
|
|
$rows
|
|
)));
|
|
}
|
|
|
|
private function databaseTableNames(mysqli $connection, string $database): array
|
|
{
|
|
$stmt = $connection->prepare(
|
|
"SELECT TABLE_NAME FROM information_schema.TABLES
|
|
WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE'
|
|
ORDER BY TABLE_NAME"
|
|
);
|
|
if ($stmt === false) {
|
|
throw new RuntimeException('Could not prepare database table comparison query.');
|
|
}
|
|
|
|
$stmt->bind_param('s', $database);
|
|
$stmt->execute();
|
|
$result = $stmt->get_result();
|
|
$rows = $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
|
|
$stmt->close();
|
|
|
|
return array_values(array_map(
|
|
static fn(array $row): string => (string)($row['TABLE_NAME'] ?? ''),
|
|
$rows
|
|
));
|
|
}
|
|
|
|
private function databaseSchemaOnlyTablesWithRows(mysqli $connection, string $database): array
|
|
{
|
|
$existingTables = [];
|
|
foreach ($this->databaseTableNames($connection, $database) as $tableName) {
|
|
$existingTables[strtolower($tableName)] = $tableName;
|
|
}
|
|
|
|
$tablesWithRows = [];
|
|
foreach (self::MARIADB_SCHEMA_ONLY_TABLES as $schemaOnlyTable) {
|
|
$actualTable = $existingTables[strtolower($schemaOnlyTable)] ?? null;
|
|
if ($actualTable === null) {
|
|
continue;
|
|
}
|
|
|
|
$row = $this->mysqliSelectOne(
|
|
$connection,
|
|
'SELECT 1 AS has_rows FROM ' . self::quoteIdentifier($database) . '.' . self::quoteIdentifier($actualTable) . ' LIMIT 1'
|
|
);
|
|
if (($row['has_rows'] ?? null) !== null) {
|
|
$tablesWithRows[] = $actualTable;
|
|
}
|
|
}
|
|
|
|
return $tablesWithRows;
|
|
}
|
|
|
|
private static function missingDatabaseTablesBlocker(string $database, array $missingTables): string
|
|
{
|
|
$shownTables = array_slice($missingTables, 0, 3);
|
|
$qualifiedTables = array_map(
|
|
static fn(string $table): string => $database . '.' . $table,
|
|
$shownTables
|
|
);
|
|
$tableWord = count($missingTables) === 1 ? 'table' : 'tables';
|
|
$sample = $qualifiedTables !== [] ? ': ' . implode(', ', $qualifiedTables) : '';
|
|
|
|
return 'Replica seed is incomplete. Missing ' . count($missingTables) . ' database ' . $tableWord . ' on replica' . $sample . '.';
|
|
}
|
|
|
|
private static function schemaOnlyTablesContainRowsBlocker(string $database, array $tables): string
|
|
{
|
|
$shownTables = array_slice($tables, 0, 3);
|
|
$qualifiedTables = array_map(
|
|
static fn(string $table): string => $database . '.' . $table,
|
|
$shownTables
|
|
);
|
|
$sample = $qualifiedTables !== [] ? ': ' . implode(', ', $qualifiedTables) : '';
|
|
|
|
return 'Replica seed includes data for schema-only tables' . $sample . '. Re-run provisioning to rebuild the replica without operational log data.';
|
|
}
|
|
|
|
private function redisClient(array $host): PredisClient
|
|
{
|
|
$credentials = $this->credentials($host);
|
|
$params = [
|
|
'scheme' => 'tcp',
|
|
'host' => (string)$host['host'],
|
|
'port' => (int)$host['port'],
|
|
'database' => (int)($host['database_index'] ?? 0),
|
|
'password' => $credentials['password'],
|
|
];
|
|
|
|
if (($credentials['username'] ?? '') !== '' && $credentials['username'] !== 'default') {
|
|
$params['username'] = $credentials['username'];
|
|
}
|
|
|
|
return new PredisClient($params);
|
|
}
|
|
|
|
private function redisInfo(PredisClient $client): array
|
|
{
|
|
$info = $client->info('replication');
|
|
if (is_array($info)) {
|
|
return isset($info['Replication']) && is_array($info['Replication'])
|
|
? $info['Replication']
|
|
: $info;
|
|
}
|
|
|
|
$parsed = [];
|
|
foreach (explode("\n", (string)$info) as $line) {
|
|
$line = trim($line);
|
|
if ($line === '' || str_starts_with($line, '#') || !str_contains($line, ':')) {
|
|
continue;
|
|
}
|
|
[$key, $value] = explode(':', $line, 2);
|
|
$parsed[$key] = trim($value);
|
|
}
|
|
return $parsed;
|
|
}
|
|
|
|
private function mysqliExec(mysqli $connection, string $sql): void
|
|
{
|
|
$connection->query($sql);
|
|
}
|
|
|
|
private function mysqliSelectOne(mysqli $connection, string $sql): array
|
|
{
|
|
$result = $connection->query($sql);
|
|
if ($result === false) {
|
|
return [];
|
|
}
|
|
$row = $result->fetch_assoc();
|
|
return is_array($row) ? $row : [];
|
|
}
|
|
|
|
private function refreshStatuses(): void
|
|
{
|
|
foreach ($this->listHosts() as $host) {
|
|
if ($this->activeOperation((string)$host['kind'], (int)$host['id']) !== null) {
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
$status = $host['kind'] === self::KIND_DATABASE
|
|
? $this->testDatabaseHost($host)
|
|
: $this->testRedisHost($host);
|
|
$this->storeStatus($host, $status);
|
|
} catch (Throwable $throwable) {
|
|
$this->storeStatus($host, [
|
|
'status' => 'degraded',
|
|
'replication_percent' => 0,
|
|
'lag_seconds' => null,
|
|
'blockers' => [$throwable->getMessage()],
|
|
'raw' => [],
|
|
'checked_at' => date('c'),
|
|
]);
|
|
}
|
|
}
|
|
}
|
|
|
|
private function storeStatus(array $host, array $status): void
|
|
{
|
|
$publicStatus = [
|
|
'status' => (string)($status['status'] ?? 'unknown'),
|
|
'replication_percent' => round((float)($status['replication_percent'] ?? 0), 2),
|
|
'lag_seconds' => $status['lag_seconds'] ?? null,
|
|
'blockers' => array_values(array_filter($status['blockers'] ?? [])),
|
|
'raw' => $status['raw'] ?? [],
|
|
'checked_at' => (string)($status['checked_at'] ?? date('c')),
|
|
];
|
|
|
|
$this->execute(
|
|
"UPDATE replication_hosts SET status = ?, last_status_json = ?, last_checked_at = NOW() WHERE id = ?",
|
|
'ssi',
|
|
[$publicStatus['status'], self::jsonEncode($publicStatus), (int)$host['id']]
|
|
);
|
|
$this->execute(
|
|
"INSERT INTO replication_status_snapshots
|
|
(host_id, kind, status, replication_percent, lag_seconds, blockers_json, raw_status_json)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
'issdiss',
|
|
[
|
|
(int)$host['id'],
|
|
(string)$host['kind'],
|
|
$publicStatus['status'],
|
|
(float)$publicStatus['replication_percent'],
|
|
$publicStatus['lag_seconds'],
|
|
self::jsonEncode($publicStatus['blockers']),
|
|
self::jsonEncode($publicStatus['raw']),
|
|
]
|
|
);
|
|
}
|
|
|
|
private function buildReplicationSummary(string $kind, array $hosts): array
|
|
{
|
|
$replicas = [];
|
|
$blockers = [];
|
|
$percents = [];
|
|
$statuses = [];
|
|
|
|
foreach ($hosts as $host) {
|
|
if (($host['role'] ?? '') === 'primary') {
|
|
continue;
|
|
}
|
|
|
|
$public = $this->publicHost($host);
|
|
$replicas[] = $public;
|
|
$status = is_array($public['last_status'] ?? null) ? $public['last_status'] : [];
|
|
$activeProgress = $public['active_operation']['progress_percent'] ?? null;
|
|
$percent = is_numeric($activeProgress) && (float)$activeProgress > 0
|
|
? (float)$activeProgress
|
|
: (float)($status['replication_percent'] ?? $public['replication_percent'] ?? 0);
|
|
$percents[] = $percent;
|
|
$statuses[] = (string)($status['status'] ?? $public['status'] ?? 'unknown');
|
|
foreach ($status['blockers'] ?? [] as $blocker) {
|
|
$blockers[] = (string)$blocker;
|
|
}
|
|
}
|
|
|
|
if ($replicas === []) {
|
|
return [
|
|
'status' => 'not_configured',
|
|
'min_percent' => 0.0,
|
|
'average_percent' => 0.0,
|
|
'replicas' => [],
|
|
'blockers' => ['No ' . $kind . ' replicas configured.'],
|
|
];
|
|
}
|
|
|
|
$min = min($percents);
|
|
$average = array_sum($percents) / max(1, count($percents));
|
|
$status = ($min >= 100.0 && $blockers === []) ? 'ok' : 'degraded';
|
|
if (in_array('down', $statuses, true)) {
|
|
$status = 'down';
|
|
}
|
|
|
|
return [
|
|
'status' => $status,
|
|
'min_percent' => round($min, 2),
|
|
'average_percent' => round($average, 2),
|
|
'replicas' => $replicas,
|
|
'blockers' => array_values(array_unique($blockers)),
|
|
];
|
|
}
|
|
|
|
private function publicHost(?array $host): ?array
|
|
{
|
|
if ($host === null) {
|
|
return null;
|
|
}
|
|
|
|
$lastStatus = self::jsonDecode($host['last_status_json'] ?? null);
|
|
$credentials = $this->credentials($host);
|
|
$activeOperation = isset($host['id'])
|
|
? $this->activeOperation((string)$host['kind'], (int)$host['id'])
|
|
: null;
|
|
return [
|
|
'id' => (int)$host['id'],
|
|
'kind' => (string)$host['kind'],
|
|
'label' => (string)$host['label'],
|
|
'host' => (string)$host['host'],
|
|
'port' => (int)$host['port'],
|
|
'database' => $host['kind'] === self::KIND_DATABASE ? (string)($host['database_name'] ?? '') : (int)($host['database_index'] ?? 0),
|
|
'role' => (string)$host['role'],
|
|
'status' => (string)$host['status'],
|
|
'replication_source_id' => isset($host['replication_source_id']) ? (int)$host['replication_source_id'] : null,
|
|
'ssl_mode' => $host['ssl_mode'] ?? null,
|
|
'replication_percent' => round((float)($lastStatus['replication_percent'] ?? (($host['role'] ?? '') === 'primary' ? 100 : 0)), 2),
|
|
'last_status' => $lastStatus,
|
|
'active_operation' => $activeOperation,
|
|
'last_checked_at' => $host['last_checked_at'] ?? null,
|
|
'credential_summary' => [
|
|
'username' => $credentials['username'] !== '' ? replication_secret_box::mask($credentials['username']) : '',
|
|
'password_set' => $credentials['password'] !== '',
|
|
'admin_username' => $credentials['admin_username'] !== '' ? replication_secret_box::mask($credentials['admin_username']) : '',
|
|
'admin_password_set' => $credentials['admin_password'] !== '',
|
|
'replication_username' => $credentials['replication_username'] !== '' ? replication_secret_box::mask($credentials['replication_username']) : '',
|
|
'replication_password_set' => $credentials['replication_password'] !== '',
|
|
],
|
|
'created_at' => $host['created_at'] ?? null,
|
|
'updated_at' => $host['updated_at'] ?? null,
|
|
'deleted_at' => $host['deleted_at'] ?? null,
|
|
];
|
|
}
|
|
|
|
private function normalizeHostInput(string $kind, array $input): array
|
|
{
|
|
$host = trim((string)($input['host'] ?? ''));
|
|
if ($host === '') {
|
|
throw new RuntimeException('Host is required.');
|
|
}
|
|
|
|
$port = (int)($input['port'] ?? ($kind === self::KIND_DATABASE ? 3306 : 6379));
|
|
if ($port <= 0 || $port > 65535) {
|
|
throw new RuntimeException('Port must be between 1 and 65535.');
|
|
}
|
|
|
|
$label = trim((string)($input['label'] ?? ''));
|
|
if ($label === '') {
|
|
$label = $host . ':' . $port;
|
|
}
|
|
|
|
$username = trim((string)($input['username'] ?? $input['user'] ?? ''));
|
|
$databaseName = null;
|
|
$databaseIndex = null;
|
|
if ($kind === self::KIND_DATABASE) {
|
|
$databaseName = trim((string)($input['database'] ?? $input['database_name'] ?? ''));
|
|
if ($databaseName === '' || $username === '') {
|
|
throw new RuntimeException('Database name and username are required for database replication hosts.');
|
|
}
|
|
} else {
|
|
$databaseIndex = (int)($input['database'] ?? $input['database_index'] ?? 0);
|
|
if ($databaseIndex < 0) {
|
|
throw new RuntimeException('Redis database index must be zero or greater.');
|
|
}
|
|
}
|
|
|
|
$options = is_array($input['options'] ?? null) ? $input['options'] : [];
|
|
|
|
return [
|
|
'label' => $label,
|
|
'host' => $host,
|
|
'port' => $port,
|
|
'database_name' => $databaseName,
|
|
'database_index' => $databaseIndex,
|
|
'username' => $username,
|
|
'password_secret' => replication_secret_box::encrypt((string)($input['password'] ?? '')),
|
|
'admin_username' => trim((string)($input['admin_username'] ?? '')),
|
|
'admin_password_secret' => replication_secret_box::encrypt((string)($input['admin_password'] ?? '')),
|
|
'replication_username' => trim((string)($input['replication_username'] ?? '')),
|
|
'replication_password_secret' => replication_secret_box::encrypt((string)($input['replication_password'] ?? '')),
|
|
'ssl_mode' => strtoupper(trim((string)($input['ssl_mode'] ?? 'DISABLED'))) ?: 'DISABLED',
|
|
'options' => $options,
|
|
];
|
|
}
|
|
|
|
private function transientHost(string $kind, array $input): array
|
|
{
|
|
$normalized = $this->normalizeHostInput($kind, $input);
|
|
$role = strtolower(trim((string)($input['role'] ?? 'replica')));
|
|
if (!in_array($role, ['primary', 'replica'], true)) {
|
|
$role = 'replica';
|
|
}
|
|
|
|
return array_merge($normalized, [
|
|
'id' => 0,
|
|
'kind' => $kind,
|
|
'role' => $role,
|
|
'status' => 'unknown',
|
|
'replication_source_id' => null,
|
|
'last_status_json' => null,
|
|
'last_checked_at' => null,
|
|
'created_at' => null,
|
|
'updated_at' => null,
|
|
'deleted_at' => null,
|
|
'test_connectivity_only' => true,
|
|
]);
|
|
}
|
|
|
|
private function ensureEnvironmentPrimaryRows(): void
|
|
{
|
|
if ($this->primaryHost(self::KIND_DATABASE) === null && isset($GLOBALS['CONFIG_DB']) && is_array($GLOBALS['CONFIG_DB'])) {
|
|
$config = $GLOBALS['CONFIG_DB'];
|
|
if (!empty($config['host']) && !empty($config['database']) && !empty($config['user'])) {
|
|
$this->insertEnvironmentPrimary(self::KIND_DATABASE, [
|
|
'label' => 'Current database primary',
|
|
'host' => (string)$config['host'],
|
|
'port' => (int)($config['port'] ?? 3306),
|
|
'database_name' => (string)$config['database'],
|
|
'database_index' => null,
|
|
'username' => (string)$config['user'],
|
|
'password_secret' => replication_secret_box::encrypt((string)($config['password'] ?? '')),
|
|
'ssl_mode' => (string)($config['ssl_mode'] ?? 'DISABLED'),
|
|
]);
|
|
}
|
|
}
|
|
|
|
if ($this->primaryHost(self::KIND_REDIS) === null && isset($GLOBALS['REDIS_CONFIG']) && is_array($GLOBALS['REDIS_CONFIG'])) {
|
|
$config = $GLOBALS['REDIS_CONFIG'];
|
|
if (!empty($config['host'])) {
|
|
$this->insertEnvironmentPrimary(self::KIND_REDIS, [
|
|
'label' => 'Current Redis primary',
|
|
'host' => (string)$config['host'],
|
|
'port' => (int)($config['port'] ?? 6379),
|
|
'database_name' => null,
|
|
'database_index' => (int)($config['database'] ?? 0),
|
|
'username' => (string)($config['user'] ?? ''),
|
|
'password_secret' => replication_secret_box::encrypt((string)($config['password'] ?? '')),
|
|
'ssl_mode' => null,
|
|
]);
|
|
}
|
|
}
|
|
}
|
|
|
|
private function insertEnvironmentPrimary(string $kind, array $host): void
|
|
{
|
|
$this->execute(
|
|
"INSERT INTO replication_hosts (
|
|
kind, label, host, port, database_name, database_index, username, password_secret,
|
|
role, status, ssl_mode, options_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'primary', 'unknown', ?, ?)",
|
|
'sssissssss',
|
|
[
|
|
$kind,
|
|
$host['label'],
|
|
$host['host'],
|
|
$host['port'],
|
|
$host['database_name'],
|
|
$host['database_index'],
|
|
$host['username'],
|
|
$host['password_secret'],
|
|
$host['ssl_mode'],
|
|
self::jsonEncode(['source' => 'environment']),
|
|
]
|
|
);
|
|
}
|
|
|
|
private function writeBootstrapSnapshot(): void
|
|
{
|
|
$databasePrimary = $this->primaryHost(self::KIND_DATABASE);
|
|
$redisPrimary = $this->primaryHost(self::KIND_REDIS);
|
|
$active = [];
|
|
|
|
if ($databasePrimary !== null) {
|
|
$credentials = $this->credentials($databasePrimary);
|
|
$active['database'] = [
|
|
'host' => (string)$databasePrimary['host'],
|
|
'port' => (int)$databasePrimary['port'],
|
|
'database' => (string)$databasePrimary['database_name'],
|
|
'user' => $credentials['username'],
|
|
'password_secret' => $databasePrimary['password_secret'] ?? '',
|
|
'ssl_mode' => (string)($databasePrimary['ssl_mode'] ?? 'DISABLED'),
|
|
];
|
|
}
|
|
|
|
if ($redisPrimary !== null) {
|
|
$credentials = $this->credentials($redisPrimary);
|
|
$active['redis'] = [
|
|
'host' => (string)$redisPrimary['host'],
|
|
'port' => (int)$redisPrimary['port'],
|
|
'database' => (int)($redisPrimary['database_index'] ?? 0),
|
|
'user' => $credentials['username'],
|
|
'password_secret' => $redisPrimary['password_secret'] ?? '',
|
|
];
|
|
}
|
|
|
|
replication_bootstrap_config::writeSnapshot([
|
|
'version' => 1,
|
|
'generated_at' => date('c'),
|
|
'active' => $active,
|
|
]);
|
|
}
|
|
|
|
private function switchPrimary(string $kind, int $newPrimaryId, int $oldPrimaryId): void
|
|
{
|
|
$this->execute(
|
|
"UPDATE replication_hosts SET role = 'inactive', status = 'inactive' WHERE kind = ? AND role = 'primary' AND id <> ?",
|
|
'si',
|
|
[$kind, $newPrimaryId]
|
|
);
|
|
$this->execute(
|
|
"UPDATE replication_hosts SET role = 'primary', status = 'ok', replication_source_id = NULL WHERE kind = ? AND id = ?",
|
|
'si',
|
|
[$kind, $newPrimaryId]
|
|
);
|
|
$this->execute(
|
|
"UPDATE replication_hosts SET replication_source_id = ? WHERE kind = ? AND role = 'replica'",
|
|
'is',
|
|
[$newPrimaryId, $kind]
|
|
);
|
|
}
|
|
|
|
private function credentials(array $host): array
|
|
{
|
|
return [
|
|
'username' => (string)($host['username'] ?? ''),
|
|
'password' => replication_secret_box::decrypt($host['password_secret'] ?? ''),
|
|
'admin_username' => (string)($host['admin_username'] ?? ''),
|
|
'admin_password' => replication_secret_box::decrypt($host['admin_password_secret'] ?? ''),
|
|
'replication_username' => (string)($host['replication_username'] ?? ''),
|
|
'replication_password' => replication_secret_box::decrypt($host['replication_password_secret'] ?? ''),
|
|
];
|
|
}
|
|
|
|
private function decodeOptions(array $host): array
|
|
{
|
|
if (isset($host['options']) && is_array($host['options'])) {
|
|
return $host['options'];
|
|
}
|
|
|
|
return self::jsonDecode($host['options_json'] ?? null);
|
|
}
|
|
|
|
private function listHosts(?string $kind = null, bool $includeDeleted = false): array
|
|
{
|
|
$where = [];
|
|
$types = '';
|
|
$params = [];
|
|
if ($kind !== null) {
|
|
$where[] = 'kind = ?';
|
|
$types .= 's';
|
|
$params[] = $kind;
|
|
}
|
|
if (!$includeDeleted) {
|
|
$where[] = 'deleted_at IS NULL';
|
|
}
|
|
|
|
$sql = 'SELECT * FROM replication_hosts';
|
|
if ($where !== []) {
|
|
$sql .= ' WHERE ' . implode(' AND ', $where);
|
|
}
|
|
$sql .= " ORDER BY FIELD(role, 'primary', 'replica', 'inactive'), id";
|
|
|
|
return $this->selectRows($sql, $types, $params);
|
|
}
|
|
|
|
private function primaryHost(string $kind): ?array
|
|
{
|
|
return $this->selectOne(
|
|
"SELECT * FROM replication_hosts WHERE kind = ? AND role = 'primary' AND deleted_at IS NULL ORDER BY id DESC LIMIT 1",
|
|
's',
|
|
[$kind]
|
|
);
|
|
}
|
|
|
|
private function getHost(string $kind, int $id, bool $includeDeleted = false): array
|
|
{
|
|
$sql = 'SELECT * FROM replication_hosts WHERE kind = ? AND id = ?';
|
|
if (!$includeDeleted) {
|
|
$sql .= ' AND deleted_at IS NULL';
|
|
}
|
|
$host = $this->selectOne($sql . ' LIMIT 1', 'si', [$kind, $id]);
|
|
if ($host === null) {
|
|
throw new RuntimeException('Replication host was not found.');
|
|
}
|
|
return $host;
|
|
}
|
|
|
|
private function startOperation(string $kind, int $hostId, string $operation, ?int $actorUserId): int
|
|
{
|
|
$this->execute(
|
|
"INSERT INTO replication_operations (kind, host_id, operation, status, actor_user_id)
|
|
VALUES (?, ?, ?, 'running', ?)",
|
|
'sisi',
|
|
[$kind, $hostId, $operation, $actorUserId]
|
|
);
|
|
return $this->insertId();
|
|
}
|
|
|
|
private function activeOperationId(string $kind, int $hostId, string $operation): ?int
|
|
{
|
|
$operationRow = $this->selectOne(
|
|
"SELECT id FROM replication_operations
|
|
WHERE kind = ? AND host_id = ? AND operation = ? AND status = 'running'
|
|
ORDER BY id DESC
|
|
LIMIT 1",
|
|
'sis',
|
|
[$kind, $hostId, $operation]
|
|
);
|
|
|
|
return $operationRow !== null ? (int)$operationRow['id'] : null;
|
|
}
|
|
|
|
private function activeOperation(string $kind, int $hostId): ?array
|
|
{
|
|
$operation = $this->selectOne(
|
|
"SELECT id, operation, status, progress_percent, message, error_message, started_at, updated_at
|
|
FROM replication_operations
|
|
WHERE kind = ? AND host_id = ? AND status = 'running'
|
|
ORDER BY id DESC
|
|
LIMIT 1",
|
|
'si',
|
|
[$kind, $hostId]
|
|
);
|
|
|
|
if ($operation === null) {
|
|
return null;
|
|
}
|
|
|
|
return [
|
|
'id' => (int)$operation['id'],
|
|
'operation' => (string)$operation['operation'],
|
|
'status' => (string)$operation['status'],
|
|
'progress_percent' => round((float)$operation['progress_percent'], 2),
|
|
'message' => $operation['message'] ?? null,
|
|
'error_message' => $operation['error_message'] ?? null,
|
|
'started_at' => $operation['started_at'] ?? null,
|
|
'updated_at' => $operation['updated_at'] ?? null,
|
|
];
|
|
}
|
|
|
|
private function operationContext(int $operationId): array
|
|
{
|
|
$operation = $this->selectOne(
|
|
'SELECT context_json FROM replication_operations WHERE id = ? LIMIT 1',
|
|
'i',
|
|
[$operationId]
|
|
);
|
|
|
|
return self::jsonDecode($operation['context_json'] ?? null);
|
|
}
|
|
|
|
private function updateOperationProgress(int $operationId, float $progress, string $message, array $context): void
|
|
{
|
|
$this->execute(
|
|
"UPDATE replication_operations
|
|
SET progress_percent = ?, message = ?, context_json = ?
|
|
WHERE id = ?",
|
|
'dssi',
|
|
[max(0, min(100, $progress)), $message, self::jsonEncode($context), $operationId]
|
|
);
|
|
}
|
|
|
|
private function finishOperation(int $operationId, string $status, float $progress, ?string $message, array $errors): void
|
|
{
|
|
$this->execute(
|
|
"UPDATE replication_operations
|
|
SET status = ?, progress_percent = ?, message = ?, error_message = ?, completed_at = NOW()
|
|
WHERE id = ?",
|
|
'sdssi',
|
|
[$status, $progress, $message, implode("\n", $errors), $operationId]
|
|
);
|
|
}
|
|
|
|
private function audit(string $kind, ?int $hostId, string $action, ?int $actorUserId, string $severity, array $context): void
|
|
{
|
|
$this->execute(
|
|
"INSERT INTO replication_audit_logs (kind, host_id, action, actor_user_id, severity, context_json)
|
|
VALUES (?, ?, ?, ?, ?, ?)",
|
|
'sisiss',
|
|
[$kind, $hostId, $action, $actorUserId, $severity, self::jsonEncode($context)]
|
|
);
|
|
}
|
|
|
|
private function acquirePromotionLock()
|
|
{
|
|
$path = (defined('WD') ? WD : dirname(__DIR__)) . DIRECTORY_SEPARATOR . 'storage' . DIRECTORY_SEPARATOR . 'replication-promotion.lock';
|
|
$dir = dirname($path);
|
|
if (!is_dir($dir) && !mkdir($dir, 0770, true) && !is_dir($dir)) {
|
|
throw new RuntimeException('Could not create promotion lock directory.');
|
|
}
|
|
$handle = fopen($path, 'c');
|
|
if ($handle === false || !flock($handle, LOCK_EX | LOCK_NB)) {
|
|
throw new RuntimeException('Another replication promotion is already running.');
|
|
}
|
|
return $handle;
|
|
}
|
|
|
|
private function releasePromotionLock($handle): void
|
|
{
|
|
if (is_resource($handle)) {
|
|
flock($handle, LOCK_UN);
|
|
fclose($handle);
|
|
}
|
|
}
|
|
|
|
private function selectOne(string $sql, string $types = '', array $params = []): ?array
|
|
{
|
|
$rows = $this->selectRows($sql, $types, $params);
|
|
return $rows[0] ?? null;
|
|
}
|
|
|
|
private function selectRows(string $sql, string $types = '', array $params = []): array
|
|
{
|
|
global $db;
|
|
if ($types === '') {
|
|
$result = $db->query($sql);
|
|
return $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
|
|
}
|
|
|
|
$stmt = $db->prepare($sql);
|
|
if ($stmt === false) {
|
|
throw new RuntimeException('Could not prepare replication query.');
|
|
}
|
|
$stmt->bind_param($types, ...$params);
|
|
$stmt->execute();
|
|
$result = $stmt->get_result();
|
|
return $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
|
|
}
|
|
|
|
private function execute(string $sql, string $types = '', array $params = []): void
|
|
{
|
|
global $db;
|
|
if ($types === '') {
|
|
$db->query($sql);
|
|
return;
|
|
}
|
|
|
|
$stmt = $db->prepare($sql);
|
|
if ($stmt === false) {
|
|
throw new RuntimeException('Could not prepare replication statement.');
|
|
}
|
|
$stmt->bind_param($types, ...$params);
|
|
$stmt->execute();
|
|
}
|
|
|
|
private function insertId(): int
|
|
{
|
|
global $db;
|
|
return (int)$db->insert_id();
|
|
}
|
|
|
|
private static function jsonEncode(mixed $value): string
|
|
{
|
|
$json = json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
|
|
if ($json === false) {
|
|
throw new RuntimeException('Could not encode replication JSON payload.');
|
|
}
|
|
return $json;
|
|
}
|
|
|
|
private static function jsonDecode(mixed $value): array
|
|
{
|
|
if (!is_string($value) || trim($value) === '') {
|
|
return [];
|
|
}
|
|
$decoded = json_decode($value, true);
|
|
return is_array($decoded) ? $decoded : [];
|
|
}
|
|
}
|