55e7aa0a87
dartssh2's SftpFileWriter leaks chunk write failures as unhandled exceptions and never completes when the server rejects data (e.g. a full /tmp), so deploys hung and spammed the console. Upload in windowed writeBytes batches instead so errors propagate, remove the partial file, and report free space on the target directory. Every deploy flow now removes its staging files in a finally block (success, failure or cancel) and sweeps peachdeploy-*/peachweb-* leftovers older than an hour from earlier crashed runs. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2285 lines
72 KiB
Dart
2285 lines
72 KiB
Dart
import 'dart:async';
|
||
import 'dart:convert';
|
||
import 'dart:typed_data';
|
||
|
||
import 'package:dartssh2/dartssh2.dart';
|
||
import '../about/external/data/Docker.dart';
|
||
import '../about/external/data/Health.dart';
|
||
import '../about/external/data/Kafka.dart';
|
||
import '../about/external/data/Redis.dart';
|
||
import '../about/internal/application/ServerProfile.dart';
|
||
import '../about/internal/application/ServiceConfig.dart';
|
||
import '../about/internal/application/StepTracker.dart';
|
||
import 'ConnectComms.dart';
|
||
|
||
enum LogLevel { info, step, success, error, remote, warn }
|
||
|
||
class LogLine {
|
||
final LogLevel level;
|
||
final String text;
|
||
final DateTime time;
|
||
LogLine(this.level, this.text) : time = DateTime.now();
|
||
}
|
||
|
||
/// Result of a deploy attempt, surfaced to the UI.
|
||
class DeployResult {
|
||
final bool ok;
|
||
final bool canRevert;
|
||
final String message;
|
||
DeployResult(this.ok, this.canRevert, this.message);
|
||
}
|
||
|
||
/// A remote filesystem entry (for the file browser).
|
||
class RemoteEntry {
|
||
final String name;
|
||
final bool isDir;
|
||
final int size;
|
||
final int? modifyTime;
|
||
RemoteEntry(this.name, this.isDir, this.size, this.modifyTime);
|
||
}
|
||
|
||
/// Live transfer stats for an upload or download.
|
||
class TransferProgress {
|
||
final int sent;
|
||
final int total;
|
||
final double bytesPerSec;
|
||
final Duration? eta;
|
||
final bool uploading;
|
||
const TransferProgress(
|
||
this.sent,
|
||
this.total,
|
||
this.bytesPerSec,
|
||
this.eta,
|
||
this.uploading,
|
||
);
|
||
|
||
double? get fraction => total <= 0 ? null : (sent / total).clamp(0.0, 1.0);
|
||
bool get complete => total > 0 && sent >= total;
|
||
}
|
||
|
||
/// Smooths transfer speed with an exponential moving average and throttles
|
||
/// emissions to ~10/s.
|
||
class _SpeedMeter {
|
||
final _start = DateTime.now();
|
||
DateTime _lastEmit;
|
||
DateTime _lastT;
|
||
int _lastSent = 0;
|
||
double _bps = 0;
|
||
_SpeedMeter()
|
||
: _lastEmit = DateTime.now().subtract(const Duration(seconds: 1)),
|
||
_lastT = DateTime.now();
|
||
|
||
/// Updates speed; returns true if the caller should emit now.
|
||
bool tick(int sent, {bool force = false}) {
|
||
final now = DateTime.now();
|
||
final dt = now.difference(_lastT).inMicroseconds / 1e6;
|
||
if (dt >= 0.15) {
|
||
final inst = (sent - _lastSent) / dt;
|
||
_bps = _bps == 0 ? inst : _bps * 0.6 + inst * 0.4;
|
||
_lastT = now;
|
||
_lastSent = sent;
|
||
} else if (_bps == 0 && sent > 0) {
|
||
final el = now.difference(_start).inMicroseconds / 1e6;
|
||
if (el > 0) _bps = sent / el;
|
||
}
|
||
if (force || now.difference(_lastEmit).inMilliseconds >= 100) {
|
||
_lastEmit = now;
|
||
return true;
|
||
}
|
||
return false;
|
||
}
|
||
|
||
double get bps => _bps;
|
||
Duration? eta(int sent, int total) {
|
||
if (_bps <= 0 || total <= 0) return null;
|
||
final remaining = (total - sent) / _bps;
|
||
return Duration(milliseconds: (remaining * 1000).round());
|
||
}
|
||
}
|
||
|
||
/// Drives one SSH session against a server: safe jar deployment, log
|
||
/// streaming, and generic SFTP file transfer.
|
||
class SshService implements ConnectComms {
|
||
SSHClient? _client;
|
||
String? _connectedServerId;
|
||
bool _rootUser = false; // when logged in as root, skip the sudo wrapper
|
||
SSHSession? _logSession;
|
||
@override
|
||
bool get isConnected => _client != null;
|
||
|
||
// Revert state after a deploy that left a backup in place.
|
||
bool _canRevert = false;
|
||
bool _revertIsWeb = false;
|
||
bool _cancelled = false;
|
||
String? _revertTarget;
|
||
String? _revertBackup;
|
||
ServerProfile? _revertServer;
|
||
ServiceConfig? _revertService;
|
||
@override
|
||
bool get canRevert => _canRevert;
|
||
|
||
final _logController = StreamController<LogLine>.broadcast();
|
||
@override
|
||
Stream<LogLine> get logs => _logController.stream;
|
||
|
||
final _progressController = StreamController<TransferProgress?>.broadcast();
|
||
@override
|
||
Stream<TransferProgress?> get transferProgress => _progressController.stream;
|
||
|
||
final steps = StepTracker();
|
||
@override
|
||
Stream<List<DeployStep>> get stepStream => steps.stream;
|
||
String? _uploadStepKey; // step whose progress the SFTP upload feeds
|
||
|
||
void _log(LogLevel level, String text) {
|
||
for (final line in const LineSplitter().convert(text)) {
|
||
if (line.trim().isEmpty && level == LogLevel.remote) continue;
|
||
_logController.add(LogLine(level, line));
|
||
}
|
||
}
|
||
|
||
Future<void> _connect(ServerProfile s) async {
|
||
// Reuse only a live connection to the same server. A long build between
|
||
// batch deploys can let the server drop an idle connection — detect that
|
||
// (isClosed) and reconnect instead of using a dead client.
|
||
if (_client != null && _connectedServerId == s.id && !_client!.isClosed) {
|
||
return;
|
||
}
|
||
if (_client != null) disconnect();
|
||
_log(LogLevel.step, 'Connecting to ${s.username}@${s.host}:${s.port} …');
|
||
final socket = await SSHSocket.connect(
|
||
s.host,
|
||
s.port,
|
||
timeout: const Duration(seconds: 15),
|
||
);
|
||
final client = SSHClient(
|
||
socket,
|
||
username: s.username,
|
||
onPasswordRequest: () => s.password,
|
||
);
|
||
await client.authenticated;
|
||
_client = client;
|
||
_connectedServerId = s.id;
|
||
_rootUser = s.username.trim().toLowerCase() == 'root';
|
||
_log(
|
||
LogLevel.success,
|
||
'Connected ✓${_rootUser ? ' (root — running without sudo)' : ''}',
|
||
);
|
||
}
|
||
|
||
/// Public connect used by the file browser.
|
||
@override
|
||
Future<void> connect(ServerProfile s) => _connect(s);
|
||
|
||
/// Runs a command, streaming stdout/stderr to the log. Feeds the sudo
|
||
/// password over stdin when [sudo] is set. Returns the exit code.
|
||
Future<int> _run(
|
||
String command, {
|
||
bool sudo = false,
|
||
String? password,
|
||
LogLevel echo = LogLevel.remote,
|
||
}) async {
|
||
final client = _client!;
|
||
final useSudo = sudo && !_rootUser;
|
||
final cmd = useSudo
|
||
? "sudo -S -p '' bash -c ${_shQuote(command)}"
|
||
: command;
|
||
final session = await client.execute(cmd);
|
||
if (useSudo && password != null) {
|
||
session.stdin.add(Uint8List.fromList(utf8.encode('$password\n')));
|
||
}
|
||
await session.stdin.close();
|
||
await Future.wait([
|
||
session.stdout
|
||
.cast<List<int>>()
|
||
.transform(utf8.decoder)
|
||
.forEach((d) => _log(echo, d.trimRight())),
|
||
session.stderr
|
||
.cast<List<int>>()
|
||
.transform(utf8.decoder)
|
||
.forEach((d) => _log(echo, d.trimRight())),
|
||
]);
|
||
await session.done;
|
||
return session.exitCode ?? -1;
|
||
}
|
||
|
||
Future<String> _capture(
|
||
String command, {
|
||
bool sudo = false,
|
||
String? password,
|
||
}) async {
|
||
final client = _client!;
|
||
final useSudo = sudo && !_rootUser;
|
||
final cmd = useSudo
|
||
? "sudo -S -p '' bash -c ${_shQuote(command)}"
|
||
: command;
|
||
final session = await client.execute(cmd);
|
||
if (useSudo && password != null) {
|
||
session.stdin.add(Uint8List.fromList(utf8.encode('$password\n')));
|
||
}
|
||
await session.stdin.close();
|
||
final out = StringBuffer();
|
||
await Future.wait([
|
||
session.stdout
|
||
.cast<List<int>>()
|
||
.transform(utf8.decoder)
|
||
.forEach(out.write),
|
||
session.stderr.cast<List<int>>().transform(utf8.decoder).drain<void>(),
|
||
]);
|
||
await session.done;
|
||
return out.toString().trim();
|
||
}
|
||
|
||
static String _shQuote(String s) => "'${s.replaceAll("'", "'\\''")}'";
|
||
|
||
// ---- jar deployment -------------------------------------------------------
|
||
|
||
@override
|
||
Future<DeployResult> deploy({
|
||
required ServerProfile server,
|
||
required ServiceConfig service,
|
||
required Uint8List jarBytes,
|
||
required String jarLabel,
|
||
}) async {
|
||
_canRevert = false;
|
||
_revertIsWeb = false;
|
||
_cancelled = false;
|
||
final sudo = server.useSudo;
|
||
final pw = server.password;
|
||
final hasService = service.serviceName.trim().isNotEmpty;
|
||
steps.reset([
|
||
('connect', 'Connect'),
|
||
('upload', 'Upload jar'),
|
||
('stop', 'Stop service'),
|
||
('backup', 'Back up current'),
|
||
('install', 'Install jar'),
|
||
('start', 'Start & verify'),
|
||
('cleanup', 'Finalize'),
|
||
('logs', 'Fetch logs'),
|
||
]);
|
||
final staging =
|
||
'/tmp/peachdeploy-${DateTime.now().millisecondsSinceEpoch}.jar';
|
||
try {
|
||
steps.active('connect');
|
||
await _connect(server);
|
||
steps.done('connect');
|
||
|
||
final target = service.remotePath(jarLabel);
|
||
final backup = '$target.bak';
|
||
final dir = target.substring(0, target.lastIndexOf('/'));
|
||
|
||
await _run('mkdir -p ${_shQuote(dir)}', sudo: sudo, password: pw);
|
||
|
||
// 1. Upload to /tmp first — the live jar is untouched until success.
|
||
steps.active('upload');
|
||
_uploadStepKey = 'upload';
|
||
_log(
|
||
LogLevel.step,
|
||
'Uploading $jarLabel (${_fmtBytes(jarBytes.length)}) → $staging',
|
||
);
|
||
await _sftpUpload(jarBytes, staging);
|
||
_uploadStepKey = null;
|
||
steps.done('upload');
|
||
_log(LogLevel.success, 'Upload complete ✓');
|
||
|
||
// 2. Stop the service before swapping.
|
||
steps.active('stop');
|
||
if (hasService) {
|
||
_log(LogLevel.step, 'Stopping service "${service.serviceName}" …');
|
||
await _run(
|
||
'systemctl stop ${_shQuote(service.serviceName)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(LogLevel.success, 'Service stopped ✓');
|
||
steps.done('stop');
|
||
} else {
|
||
steps.skip('stop');
|
||
}
|
||
|
||
// 3. Back up the current jar to <name>.bak.
|
||
steps.active('backup');
|
||
_log(LogLevel.step, 'Backing up current jar → $backup');
|
||
await _run(
|
||
'rm -f ${_shQuote(backup)}; '
|
||
'if [ -f ${_shQuote(target)} ]; then mv ${_shQuote(target)} ${_shQuote(backup)}; '
|
||
'echo "backed up"; else echo "no previous jar"; fi',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
steps.done('backup');
|
||
|
||
// 4. Install the new jar under the required name.
|
||
steps.active('install');
|
||
_log(LogLevel.step, 'Installing → $target');
|
||
await _run(
|
||
'mv -f ${_shQuote(staging)} ${_shQuote(target)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
if (service.isGeneric) {
|
||
await _run('chmod +x ${_shQuote(target)}', sudo: sudo, password: pw);
|
||
}
|
||
// Verify the artifact actually landed (SFTP/mv can fail on a dropped
|
||
// connection during a long batch).
|
||
final placed = await _capture(
|
||
'if [ -f ${_shQuote(target)} ]; then echo "OK \$(wc -c < ${_shQuote(target)})"; else echo MISSING; fi',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(LogLevel.remote, 'artifact check: $placed');
|
||
if (placed.contains('MISSING')) {
|
||
steps.failRemaining();
|
||
// Restore the previous jar so the service isn't left broken.
|
||
await _run(
|
||
'if [ -f ${_shQuote(backup)} ]; then mv -f ${_shQuote(backup)} ${_shQuote(target)}; fi',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(LogLevel.error, 'Artifact not present at $target after install.');
|
||
return DeployResult(
|
||
false,
|
||
false,
|
||
'Install failed — jar not present at $target.',
|
||
);
|
||
}
|
||
steps.done('install');
|
||
_log(LogLevel.success, 'New artifact installed ✓');
|
||
|
||
// 5. Start and verify.
|
||
steps.active('start');
|
||
var ok = true;
|
||
if (hasService) {
|
||
_log(LogLevel.step, 'Starting service "${service.serviceName}" …');
|
||
await _run(
|
||
'systemctl start ${_shQuote(service.serviceName)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
await Future.delayed(const Duration(seconds: 2));
|
||
final state = await _capture(
|
||
'systemctl is-active ${_shQuote(service.serviceName)} || true',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
ok = state.contains('active') && !state.contains('inactive');
|
||
if (ok) {
|
||
_log(LogLevel.success, 'Service is active ✓');
|
||
} else {
|
||
_log(LogLevel.error, 'Service is "$state" — start FAILED.');
|
||
await _run(
|
||
'systemctl --no-pager status ${_shQuote(service.serviceName)} | head -n 12',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
}
|
||
}
|
||
ok ? steps.done('start') : steps.fail('start');
|
||
|
||
steps.active('cleanup');
|
||
if (ok) {
|
||
await _run('rm -f ${_shQuote(backup)}', sudo: sudo, password: pw);
|
||
_log(LogLevel.success, 'Deployment successful 🎉');
|
||
_canRevert = false;
|
||
steps.done('cleanup');
|
||
} else {
|
||
_canRevert = true;
|
||
_revertTarget = target;
|
||
_revertBackup = backup;
|
||
_revertServer = server;
|
||
_revertService = service;
|
||
_log(
|
||
LogLevel.warn,
|
||
'Backup preserved at $backup — you can revert to the previous jar.',
|
||
);
|
||
steps.skip('cleanup');
|
||
}
|
||
|
||
steps.active('logs');
|
||
_log(LogLevel.step, 'Fetching recent logs …');
|
||
await _runLogCommand(server, service);
|
||
steps.done('logs');
|
||
|
||
return DeployResult(
|
||
ok,
|
||
_canRevert,
|
||
ok
|
||
? 'Deployed $jarLabel and service is active.'
|
||
: 'Service failed to start.',
|
||
);
|
||
} catch (e) {
|
||
steps.failRemaining();
|
||
if (_cancelled) {
|
||
_log(LogLevel.warn, 'Deployment cancelled by user.');
|
||
return DeployResult(false, _canRevert, 'Deployment cancelled.');
|
||
}
|
||
_log(LogLevel.error, 'Deploy failed: $e');
|
||
return DeployResult(false, _canRevert, 'Error: $e');
|
||
} finally {
|
||
await _cleanStaging(server, [staging]);
|
||
}
|
||
}
|
||
|
||
// ---- web (nginx) deployment ----------------------------------------------
|
||
|
||
/// Deploys a static site (Angular/React/Flutter web/static) to the nginx web
|
||
/// root. [archiveBytes] is a .tar.gz of the built site's contents.
|
||
@override
|
||
Future<DeployResult> deployWeb({
|
||
required ServerProfile server,
|
||
required ServiceConfig service,
|
||
required Uint8List archiveBytes,
|
||
required int fileCount,
|
||
}) async {
|
||
_canRevert = false;
|
||
_revertIsWeb = true;
|
||
_cancelled = false;
|
||
final sudo = server.useSudo;
|
||
final pw = server.password;
|
||
final root = service.remoteRoot;
|
||
final backup = '$root.bak';
|
||
final nginx = service.nginxService.trim().isEmpty
|
||
? 'nginx'
|
||
: service.nginxService.trim();
|
||
final ts = DateTime.now().millisecondsSinceEpoch;
|
||
final archive = '/tmp/peachdeploy-web-$ts.tgz';
|
||
final staging = '/tmp/peachweb-$ts'; // extract & verify here first
|
||
steps.reset([
|
||
('connect', 'Connect'),
|
||
('upload', 'Upload bundle'),
|
||
('extract', 'Extract & verify'),
|
||
('backup', 'Back up web root'),
|
||
('publish', 'Publish $fileCount files'),
|
||
('reload', 'Reload nginx'),
|
||
('cleanup', 'Finalize'),
|
||
('logs', 'Fetch logs'),
|
||
]);
|
||
try {
|
||
steps.active('connect');
|
||
await _connect(server);
|
||
steps.done('connect');
|
||
|
||
// Ensure the web root (e.g. /var/www/<project>) exists — create it and
|
||
// any parent folders if missing.
|
||
_log(LogLevel.step, 'Ensuring web root $root exists …');
|
||
await _run('mkdir -p ${_shQuote(root)}', sudo: sudo, password: pw);
|
||
|
||
// 1. Upload the bundle to /tmp.
|
||
steps.active('upload');
|
||
_uploadStepKey = 'upload';
|
||
_log(
|
||
LogLevel.step,
|
||
'Uploading site bundle (${_fmtBytes(archiveBytes.length)}, $fileCount files) …',
|
||
);
|
||
await _sftpUpload(archiveBytes, archive);
|
||
_uploadStepKey = null;
|
||
steps.done('upload');
|
||
_log(LogLevel.success, 'Upload complete ✓');
|
||
|
||
// 2. Extract to a STAGING dir and verify BEFORE touching the live root,
|
||
// so a bad/empty bundle can never wipe the site.
|
||
steps.active('extract');
|
||
_log(LogLevel.step, 'Extracting bundle to $staging …');
|
||
await _run(
|
||
'rm -rf ${_shQuote(staging)} && mkdir -p ${_shQuote(staging)} && '
|
||
'tar -xzf ${_shQuote(archive)} -C ${_shQuote(staging)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
final verify = await _capture(
|
||
'if [ -f ${_shQuote('$staging/index.html')} ]; then '
|
||
'echo "OK \$(find ${_shQuote(staging)} -type f | wc -l)"; '
|
||
'else echo "MISSING \$(find ${_shQuote(staging)} -type f | wc -l)"; fi',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(LogLevel.remote, 'staging check: $verify');
|
||
if (!verify.contains('OK')) {
|
||
steps.failRemaining();
|
||
await _run(
|
||
'rm -rf ${_shQuote(staging)} ${_shQuote(archive)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(
|
||
LogLevel.error,
|
||
'No index.html in the bundle — the LIVE site at $root was NOT touched. '
|
||
'Check the build output folder (must be the folder that contains index.html, '
|
||
'e.g. dist/<project>/browser).',
|
||
);
|
||
return DeployResult(
|
||
false,
|
||
false,
|
||
'Bundle has no index.html — nothing published (site untouched).',
|
||
);
|
||
}
|
||
steps.done('extract');
|
||
|
||
// 3. Back up the current web root (only now that we know staging is good).
|
||
steps.active('backup');
|
||
_log(LogLevel.step, 'Backing up web root → $backup');
|
||
await _run(
|
||
'mkdir -p ${_shQuote(root)}; rm -rf ${_shQuote(backup)}; '
|
||
'if [ -d ${_shQuote(root)} ] && [ -n "\$(ls -A ${_shQuote(root)} 2>/dev/null)" ]; '
|
||
'then cp -a ${_shQuote(root)} ${_shQuote(backup)}; fi',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
steps.done('backup');
|
||
|
||
// 4. Atomic-ish swap: clear the root and copy staging contents in.
|
||
steps.active('publish');
|
||
_log(LogLevel.step, 'Publishing new site to $root …');
|
||
final publishCode = await _run(
|
||
'find ${_shQuote(root)} -mindepth 1 -delete && '
|
||
'cp -a ${_shQuote('$staging/.')} ${_shQuote(root)}/ && '
|
||
'rm -rf ${_shQuote(staging)} ${_shQuote(archive)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
final live = await _capture(
|
||
'if [ -f ${_shQuote('$root/index.html')} ]; then echo "OK \$(find ${_shQuote(root)} -type f | wc -l)"; else echo MISSING; fi',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(LogLevel.remote, 'live check: $live');
|
||
if (publishCode != 0 || live.contains('MISSING')) {
|
||
// Restore the backup so the site is never left empty.
|
||
_log(LogLevel.warn, 'Publish issue — restoring previous site …');
|
||
await _run(
|
||
'find ${_shQuote(root)} -mindepth 1 -delete; '
|
||
'if [ -d ${_shQuote(backup)} ]; then cp -a ${_shQuote('$backup/.')} ${_shQuote(root)}/; fi',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
steps.failRemaining();
|
||
return DeployResult(
|
||
false,
|
||
false,
|
||
'Publish failed — restored the previous site.',
|
||
);
|
||
}
|
||
steps.done('publish');
|
||
final listing = await _capture(
|
||
'ls -la ${_shQuote(root)} | head -n 25',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(LogLevel.success, 'Published to $root:');
|
||
_log(LogLevel.remote, listing);
|
||
|
||
// 4. Test config and reload nginx.
|
||
steps.active('reload');
|
||
var ok = true;
|
||
_log(LogLevel.step, 'Testing nginx config & reloading …');
|
||
final test = await _capture(
|
||
'nginx -t 2>&1 || true',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(LogLevel.remote, test);
|
||
if (test.contains('test failed') || test.contains('emerg')) {
|
||
ok = false;
|
||
_log(LogLevel.error, 'nginx config test FAILED.');
|
||
} else {
|
||
await _run(
|
||
'systemctl reload ${_shQuote(nginx)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
await Future.delayed(const Duration(seconds: 1));
|
||
final state = await _capture(
|
||
'systemctl is-active ${_shQuote(nginx)} || true',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
ok = state.contains('active') && !state.contains('inactive');
|
||
_log(
|
||
ok ? LogLevel.success : LogLevel.error,
|
||
ok ? 'nginx reloaded ✓' : 'nginx is "$state".',
|
||
);
|
||
}
|
||
ok ? steps.done('reload') : steps.fail('reload');
|
||
|
||
steps.active('cleanup');
|
||
if (ok) {
|
||
await _run('rm -rf ${_shQuote(backup)}', sudo: sudo, password: pw);
|
||
_log(LogLevel.success, 'Website deployed 🎉');
|
||
_canRevert = false;
|
||
steps.done('cleanup');
|
||
} else {
|
||
_canRevert = true;
|
||
_revertTarget = root;
|
||
_revertBackup = backup;
|
||
_revertServer = server;
|
||
_revertService = service;
|
||
_log(LogLevel.warn, 'Previous site kept at $backup — you can revert.');
|
||
steps.skip('cleanup');
|
||
}
|
||
|
||
steps.active('logs');
|
||
await _runLogCommand(server, service);
|
||
steps.done('logs');
|
||
|
||
return DeployResult(
|
||
ok,
|
||
_canRevert,
|
||
ok ? 'Published $fileCount files to $root.' : 'nginx failed to reload.',
|
||
);
|
||
} catch (e) {
|
||
steps.failRemaining();
|
||
if (_cancelled) {
|
||
_log(LogLevel.warn, 'Deployment cancelled by user.');
|
||
return DeployResult(false, _canRevert, 'Deployment cancelled.');
|
||
}
|
||
_log(LogLevel.error, 'Web deploy failed: $e');
|
||
return DeployResult(false, _canRevert, 'Error: $e');
|
||
} finally {
|
||
await _cleanStaging(server, [archive, staging]);
|
||
}
|
||
}
|
||
|
||
/// Best-effort cancel of an in-flight deploy: interrupts the current command
|
||
/// by dropping the connection. Steps already applied on the server are not
|
||
/// rolled back automatically.
|
||
@override
|
||
void requestCancel() {
|
||
_cancelled = true;
|
||
_log(LogLevel.warn, 'Cancelling — closing connection …');
|
||
try {
|
||
_client?.close();
|
||
} catch (_) {}
|
||
_client = null;
|
||
_connectedServerId = null;
|
||
}
|
||
|
||
/// Restarts a service (or reloads nginx for web apps) without deploying,
|
||
/// then shows recent logs. Returns success.
|
||
@override
|
||
Future<DeployResult> restartOnly(
|
||
ServerProfile server,
|
||
ServiceConfig service,
|
||
) async {
|
||
final sudo = server.useSudo;
|
||
final pw = server.password;
|
||
try {
|
||
await _connect(server);
|
||
if (service.isWeb) {
|
||
final nginx = service.nginxService.trim().isEmpty
|
||
? 'nginx'
|
||
: service.nginxService.trim();
|
||
_log(LogLevel.step, 'Reloading $nginx …');
|
||
await _run(
|
||
'systemctl reload ${_shQuote(nginx)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
final state = await _capture(
|
||
'systemctl is-active ${_shQuote(nginx)} || true',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
final ok = state.contains('active') && !state.contains('inactive');
|
||
_log(
|
||
ok ? LogLevel.success : LogLevel.error,
|
||
ok ? 'nginx reloaded ✓' : 'nginx is "$state".',
|
||
);
|
||
await _runLogCommand(server, service);
|
||
return DeployResult(
|
||
ok,
|
||
false,
|
||
ok ? 'nginx reloaded.' : 'Reload failed.',
|
||
);
|
||
}
|
||
final unit = service.serviceName.trim();
|
||
if (unit.isEmpty) {
|
||
return DeployResult(
|
||
false,
|
||
false,
|
||
'No systemd unit set for this service.',
|
||
);
|
||
}
|
||
_log(LogLevel.step, 'Restarting $unit …');
|
||
await _run(
|
||
'systemctl restart ${_shQuote(unit)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
await Future.delayed(const Duration(seconds: 2));
|
||
final state = await _capture(
|
||
'systemctl is-active ${_shQuote(unit)} || true',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
final ok = state.contains('active') && !state.contains('inactive');
|
||
_log(
|
||
ok ? LogLevel.success : LogLevel.error,
|
||
ok ? 'Service is active ✓' : 'Service is "$state".',
|
||
);
|
||
await _runLogCommand(server, service);
|
||
return DeployResult(
|
||
ok,
|
||
false,
|
||
ok ? 'Service restarted.' : 'Restart failed.',
|
||
);
|
||
} catch (e) {
|
||
_log(LogLevel.error, 'Restart failed: $e');
|
||
return DeployResult(false, false, 'Error: $e');
|
||
}
|
||
}
|
||
|
||
@override
|
||
Future<DeployResult> revert() async {
|
||
final server = _revertServer;
|
||
final service = _revertService;
|
||
final target = _revertTarget;
|
||
final backup = _revertBackup;
|
||
if (server == null ||
|
||
service == null ||
|
||
target == null ||
|
||
backup == null ||
|
||
!_canRevert) {
|
||
return DeployResult(false, false, 'Nothing to revert.');
|
||
}
|
||
final sudo = server.useSudo;
|
||
final pw = server.password;
|
||
try {
|
||
if (_revertIsWeb) {
|
||
_log(LogLevel.step, 'Reverting web root to previous site …');
|
||
await _run(
|
||
'if [ -d ${_shQuote(backup)} ]; then '
|
||
'find ${_shQuote(target)} -mindepth 1 -delete; '
|
||
'cp -a ${_shQuote('$backup/.')} ${_shQuote(target)}/; '
|
||
'rm -rf ${_shQuote(backup)}; '
|
||
'else echo "no backup to restore"; fi',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
final nginx = service.nginxService.trim().isEmpty
|
||
? 'nginx'
|
||
: service.nginxService.trim();
|
||
await _run(
|
||
'systemctl reload ${_shQuote(nginx)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
_log(LogLevel.success, 'Reverted to previous site ✓');
|
||
} else {
|
||
_log(LogLevel.step, 'Reverting to previous jar …');
|
||
if (service.serviceName.trim().isNotEmpty) {
|
||
await _run(
|
||
'systemctl stop ${_shQuote(service.serviceName)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
}
|
||
await _run('rm -f ${_shQuote(target)}', sudo: sudo, password: pw);
|
||
await _run(
|
||
'mv -f ${_shQuote(backup)} ${_shQuote(target)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
if (service.serviceName.trim().isNotEmpty) {
|
||
await _run(
|
||
'systemctl start ${_shQuote(service.serviceName)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
}
|
||
_log(LogLevel.success, 'Reverted to previous jar ✓');
|
||
}
|
||
_canRevert = false;
|
||
await _runLogCommand(server, service);
|
||
return DeployResult(true, false, 'Reverted to the previous version.');
|
||
} catch (e) {
|
||
_log(LogLevel.error, 'Revert failed: $e');
|
||
return DeployResult(false, true, 'Revert error: $e');
|
||
}
|
||
}
|
||
|
||
@override
|
||
Future<void> discardBackup() async {
|
||
final server = _revertServer;
|
||
final backup = _revertBackup;
|
||
if (server == null || backup == null) return;
|
||
_log(LogLevel.step, 'Discarding backup $backup …');
|
||
final cmd = _revertIsWeb
|
||
? 'rm -rf ${_shQuote(backup)}'
|
||
: 'rm -f ${_shQuote(backup)}';
|
||
await _run(cmd, sudo: server.useSudo, password: server.password);
|
||
_canRevert = false;
|
||
_log(LogLevel.success, 'Backup discarded.');
|
||
}
|
||
|
||
// ---- logs -----------------------------------------------------------------
|
||
|
||
Future<void> _runLogCommand(
|
||
ServerProfile server,
|
||
ServiceConfig service,
|
||
) async {
|
||
await _run(
|
||
service.effectiveLogCommand,
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
echo: LogLevel.remote,
|
||
);
|
||
}
|
||
|
||
/// Writes and registers a systemd unit file: upload → move to
|
||
/// /etc/systemd/system → daemon-reload → (enable) → (start) → status.
|
||
@override
|
||
Future<DeployResult> installSystemdUnit({
|
||
required ServerProfile server,
|
||
required String unitName,
|
||
required String unitContent,
|
||
bool enable = true,
|
||
bool startNow = true,
|
||
}) async {
|
||
final sudo = server.useSudo;
|
||
final pw = server.password;
|
||
final tmp =
|
||
'/tmp/peachdeploy-$unitName-${DateTime.now().millisecondsSinceEpoch}.service';
|
||
final target = '/etc/systemd/system/$unitName.service';
|
||
steps.reset([
|
||
('connect', 'Connect'),
|
||
('upload', 'Upload unit file'),
|
||
('install', 'Install unit'),
|
||
('reload', 'daemon-reload'),
|
||
('enable', 'Enable on boot'),
|
||
('start', 'Start service'),
|
||
('status', 'Check status'),
|
||
]);
|
||
try {
|
||
steps.active('connect');
|
||
await _connect(server);
|
||
steps.done('connect');
|
||
|
||
steps.active('upload');
|
||
_log(LogLevel.step, 'Uploading unit file → $tmp');
|
||
await _sftpUpload(Uint8List.fromList(utf8.encode(unitContent)), tmp);
|
||
steps.done('upload');
|
||
|
||
steps.active('install');
|
||
_log(LogLevel.step, 'Installing → $target');
|
||
await _run(
|
||
'mv -f ${_shQuote(tmp)} ${_shQuote(target)} && chmod 644 ${_shQuote(target)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
steps.done('install');
|
||
|
||
steps.active('reload');
|
||
await _run('systemctl daemon-reload', sudo: sudo, password: pw);
|
||
steps.done('reload');
|
||
|
||
steps.active('enable');
|
||
if (enable) {
|
||
await _run(
|
||
'systemctl enable ${_shQuote(unitName)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
steps.done('enable');
|
||
} else {
|
||
steps.skip('enable');
|
||
}
|
||
|
||
steps.active('start');
|
||
if (startNow) {
|
||
await _run(
|
||
'systemctl restart ${_shQuote(unitName)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
await Future.delayed(const Duration(seconds: 2));
|
||
steps.done('start');
|
||
} else {
|
||
steps.skip('start');
|
||
}
|
||
|
||
steps.active('status');
|
||
final st = await _capture(
|
||
'systemctl is-active ${_shQuote(unitName)} || true',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
await _run(
|
||
'systemctl --no-pager status ${_shQuote(unitName)} | head -n 12',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
steps.done('status');
|
||
|
||
final ok =
|
||
!startNow || (st.contains('active') && !st.contains('inactive'));
|
||
_log(
|
||
ok ? LogLevel.success : LogLevel.warn,
|
||
'Unit $unitName installed${startNow ? ' — is "$st"' : ''}.',
|
||
);
|
||
return DeployResult(
|
||
ok,
|
||
false,
|
||
ok
|
||
? '$unitName registered${startNow ? ' and active' : ''}.'
|
||
: '$unitName installed but is "$st".',
|
||
);
|
||
} catch (e) {
|
||
steps.failRemaining();
|
||
_log(LogLevel.error, 'Unit install failed: $e');
|
||
return DeployResult(false, false, 'Error: $e');
|
||
} finally {
|
||
await _cleanStaging(server, [tmp]);
|
||
}
|
||
}
|
||
|
||
// ---- environment (configuration tab) --------------------------------------
|
||
|
||
/// Detects the server's package manager: apt | dnf | yum | apk | zypper |
|
||
/// pacman | unknown.
|
||
@override
|
||
Future<String> detectPackageManager(ServerProfile server) async {
|
||
await _connect(server);
|
||
final out = await _capture(
|
||
'command -v apt-get >/dev/null 2>&1 && echo apt; '
|
||
'command -v dnf >/dev/null 2>&1 && echo dnf; '
|
||
'command -v yum >/dev/null 2>&1 && echo yum; '
|
||
'command -v apk >/dev/null 2>&1 && echo apk; '
|
||
'command -v zypper >/dev/null 2>&1 && echo zypper; '
|
||
'command -v pacman >/dev/null 2>&1 && echo pacman',
|
||
sudo: false,
|
||
);
|
||
final lines = out
|
||
.trim()
|
||
.split('\n')
|
||
.where((l) => l.trim().isNotEmpty)
|
||
.toList();
|
||
return lines.isEmpty ? 'unknown' : lines.first.trim();
|
||
}
|
||
|
||
/// Runs each detect command; returns key -> version line ('' when missing).
|
||
@override
|
||
Future<Map<String, String>> probeEnvironment(
|
||
ServerProfile server,
|
||
List<String> detectByKey,
|
||
) async {
|
||
// detectByKey is a flat list of "key command" pairs to avoid importing
|
||
// the model here.
|
||
await _connect(server);
|
||
final res = <String, String>{};
|
||
for (final pair in detectByKey) {
|
||
final i = pair.indexOf(' ');
|
||
final key = pair.substring(0, i);
|
||
final cmd = pair.substring(i + 1);
|
||
try {
|
||
res[key] = (await _capture(cmd, sudo: false)).trim();
|
||
} catch (_) {
|
||
res[key] = '';
|
||
}
|
||
}
|
||
return res;
|
||
}
|
||
|
||
/// Installs a set of packages (one combined command) and enables each
|
||
/// listed systemd unit. Used by the new-server setup wizard.
|
||
@override
|
||
Future<DeployResult> installEnvironment({
|
||
required ServerProfile server,
|
||
required String command,
|
||
required List<String> units,
|
||
}) async {
|
||
final sudo = server.useSudo;
|
||
final pw = server.password;
|
||
steps.reset([
|
||
('connect', 'Connect'),
|
||
('install', 'Install packages'),
|
||
('enable', 'Enable services'),
|
||
('verify', 'Finalize'),
|
||
]);
|
||
try {
|
||
steps.active('connect');
|
||
await _connect(server);
|
||
steps.done('connect');
|
||
|
||
steps.active('install');
|
||
_log(LogLevel.step, '\$ $command');
|
||
final code = await _run(command, sudo: sudo, password: pw);
|
||
if (code != 0) {
|
||
steps.failRemaining();
|
||
return DeployResult(false, false, 'Install failed (exit $code).');
|
||
}
|
||
steps.done('install');
|
||
|
||
steps.active('enable');
|
||
for (final u in units.where((u) => u.trim().isNotEmpty)) {
|
||
_log(LogLevel.step, 'Enabling $u …');
|
||
await _run(
|
||
'systemctl enable --now ${_shQuote(u)} || true',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
}
|
||
steps.done('enable');
|
||
|
||
steps.active('verify');
|
||
steps.done('verify');
|
||
_log(LogLevel.success, 'Environment ready ✓');
|
||
return DeployResult(true, false, 'Environment set up.');
|
||
} catch (e) {
|
||
steps.failRemaining();
|
||
_log(LogLevel.error, 'Setup failed: $e');
|
||
return DeployResult(false, false, 'Error: $e');
|
||
}
|
||
}
|
||
|
||
/// Runs a guided install command (already confirmed by the user) and
|
||
/// optionally enables a systemd unit afterwards.
|
||
@override
|
||
Future<DeployResult> installPackages({
|
||
required ServerProfile server,
|
||
required String command,
|
||
String? enableUnit,
|
||
}) async {
|
||
final sudo = server.useSudo;
|
||
final pw = server.password;
|
||
steps.reset([
|
||
('connect', 'Connect'),
|
||
('install', 'Install packages'),
|
||
('enable', 'Enable service'),
|
||
('verify', 'Finalize'),
|
||
]);
|
||
try {
|
||
steps.active('connect');
|
||
await _connect(server);
|
||
steps.done('connect');
|
||
|
||
steps.active('install');
|
||
_log(LogLevel.step, '\$ $command');
|
||
final code = await _run(command, sudo: sudo, password: pw);
|
||
if (code != 0) {
|
||
steps.failRemaining();
|
||
_log(LogLevel.error, 'Install failed (exit $code).');
|
||
return DeployResult(false, false, 'Install failed (exit $code).');
|
||
}
|
||
steps.done('install');
|
||
|
||
steps.active('enable');
|
||
if (enableUnit != null && enableUnit.trim().isNotEmpty) {
|
||
await _run(
|
||
'systemctl enable --now ${_shQuote(enableUnit)} || true',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
steps.done('enable');
|
||
} else {
|
||
steps.skip('enable');
|
||
}
|
||
|
||
steps.active('verify');
|
||
steps.done('verify');
|
||
_log(LogLevel.success, 'Installation complete ✓');
|
||
return DeployResult(true, false, 'Installed successfully.');
|
||
} catch (e) {
|
||
steps.failRemaining();
|
||
_log(LogLevel.error, 'Install failed: $e');
|
||
return DeployResult(false, false, 'Error: $e');
|
||
}
|
||
}
|
||
|
||
/// Probes a service's health per its configured [HealthCheckType]. Quiet:
|
||
/// does not stream to the log console (used by the health dashboard).
|
||
@override
|
||
Future<HealthResult> checkHealth(
|
||
ServerProfile server,
|
||
ServiceConfig service,
|
||
) async {
|
||
final pw = server.password;
|
||
final sudo = server.useSudo;
|
||
try {
|
||
await _connect(server);
|
||
switch (service.healthType) {
|
||
case HealthCheckType.none:
|
||
return HealthResult(HealthStatus.disabled, 'disabled');
|
||
case HealthCheckType.auto:
|
||
final unit = _unitFor(service);
|
||
if (unit.isEmpty) {
|
||
return HealthResult(HealthStatus.unknown, 'no unit');
|
||
}
|
||
final out = await _capture(
|
||
'systemctl is-active ${_shQuote(unit)} || true',
|
||
sudo: false,
|
||
);
|
||
final s = out.trim().split('\n').last.trim();
|
||
final ok = s.contains('active') && !s.contains('inactive');
|
||
return HealthResult(
|
||
ok ? HealthStatus.healthy : HealthStatus.unhealthy,
|
||
s.isEmpty ? 'unknown' : s,
|
||
);
|
||
case HealthCheckType.http:
|
||
final url = service.healthTarget.trim();
|
||
if (url.isEmpty) {
|
||
return HealthResult(HealthStatus.unknown, 'no URL');
|
||
}
|
||
final out = await _capture(
|
||
'curl -s -o /dev/null -w "%{http_code}" --max-time 6 ${_shQuote(url)} 2>/dev/null || echo 000',
|
||
sudo: false,
|
||
);
|
||
final code = int.tryParse(out.trim().split('\n').last.trim()) ?? 0;
|
||
final ok = code >= 200 && code < 400;
|
||
return HealthResult(
|
||
ok ? HealthStatus.healthy : HealthStatus.unhealthy,
|
||
'HTTP $code',
|
||
);
|
||
case HealthCheckType.tcp:
|
||
final port = service.healthTarget.trim();
|
||
if (port.isEmpty) {
|
||
return HealthResult(HealthStatus.unknown, 'no port');
|
||
}
|
||
final out = await _capture(
|
||
'timeout 4 bash -c "echo > /dev/tcp/127.0.0.1/$port" >/dev/null 2>&1 && echo open || echo closed',
|
||
sudo: false,
|
||
);
|
||
final ok = out.contains('open');
|
||
return HealthResult(
|
||
ok ? HealthStatus.healthy : HealthStatus.unhealthy,
|
||
ok ? 'port $port open' : 'port $port closed',
|
||
);
|
||
case HealthCheckType.command:
|
||
final cmd = service.healthTarget.trim();
|
||
if (cmd.isEmpty) {
|
||
return HealthResult(HealthStatus.unknown, 'no command');
|
||
}
|
||
final out = await _capture(
|
||
'($cmd) >/dev/null 2>&1 && echo OK || echo FAIL',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
final ok = out.contains('OK');
|
||
return HealthResult(
|
||
ok ? HealthStatus.healthy : HealthStatus.unhealthy,
|
||
ok ? 'exit 0' : 'non-zero exit',
|
||
);
|
||
}
|
||
} catch (e) {
|
||
return HealthResult(HealthStatus.unknown, 'error: $e');
|
||
}
|
||
}
|
||
|
||
// ---- redis viewer ---------------------------------------------------------
|
||
|
||
String _redisCli({
|
||
required String host,
|
||
required int port,
|
||
required String pass,
|
||
required int db,
|
||
}) {
|
||
// `timeout 8` guards against a hung connect; --no-auth-warning only when a
|
||
// password is passed (older redis-cli rejects the flag otherwise).
|
||
final b = StringBuffer('timeout 8 redis-cli');
|
||
b.write(' -h ${_shQuote(host.trim().isEmpty ? '127.0.0.1' : host.trim())}');
|
||
b.write(' -p $port');
|
||
if (pass.trim().isNotEmpty) {
|
||
b.write(' -a ${_shQuote(pass.trim())} --no-auth-warning');
|
||
}
|
||
if (db != 0) b.write(' -n $db');
|
||
return b.toString();
|
||
}
|
||
|
||
/// (ok, detail) — verifies redis-cli exists and PING returns PONG.
|
||
@override
|
||
Future<(bool, String)> redisPing(
|
||
ServerProfile server, {
|
||
String host = '127.0.0.1',
|
||
int port = 6379,
|
||
String pass = '',
|
||
int db = 0,
|
||
}) async {
|
||
await _connect(server);
|
||
final has = await _capture(
|
||
'command -v redis-cli >/dev/null 2>&1 && echo yes || echo no',
|
||
sudo: false,
|
||
);
|
||
if (has.contains('no')) {
|
||
return (false, 'redis-cli is not installed on the server.');
|
||
}
|
||
final out = await _capture(
|
||
'${_redisCli(host: host, port: port, pass: pass, db: db)} PING 2>&1',
|
||
sudo: false,
|
||
);
|
||
final ok = out.trim().toUpperCase().contains('PONG');
|
||
return (ok, out.trim());
|
||
}
|
||
|
||
/// Reads `requirepass` from the server's Redis config (needs root/sudo to
|
||
/// read /etc/redis). Returns '' if none is set / not found.
|
||
@override
|
||
Future<String> redisConfigPassword(ServerProfile server) async {
|
||
await _connect(server);
|
||
final line = await _capture(
|
||
'for f in /etc/redis/redis.conf /etc/redis.conf /etc/redis/redis-server.conf '
|
||
'/etc/redis/6379.conf; do [ -f "\$f" ] && grep -iE "^[[:space:]]*requirepass[[:space:]]" "\$f"; done | tail -n1',
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
);
|
||
final m = RegExp(
|
||
r'requirepass\s+(.+)',
|
||
caseSensitive: false,
|
||
).firstMatch(line.trim());
|
||
if (m == null) return '';
|
||
return m.group(1)!.trim().replaceAll('"', '').replaceAll("'", '');
|
||
}
|
||
|
||
/// Total number of keys in the selected DB (`DBSIZE`).
|
||
@override
|
||
Future<int> redisDbSize(
|
||
ServerProfile server, {
|
||
String host = '127.0.0.1',
|
||
int port = 6379,
|
||
String pass = '',
|
||
int db = 0,
|
||
}) async {
|
||
await _connect(server);
|
||
final out = await _capture(
|
||
'${_redisCli(host: host, port: port, pass: pass, db: db)} DBSIZE 2>&1',
|
||
sudo: false,
|
||
);
|
||
return int.tryParse(
|
||
RegExp(r'\d+').firstMatch(out.trim())?.group(0) ?? '',
|
||
) ??
|
||
0;
|
||
}
|
||
|
||
/// Returns (number of DBs, {dbIndex: keyCount}) from CONFIG + INFO keyspace.
|
||
@override
|
||
Future<(int, Map<int, int>)> redisDatabases(
|
||
ServerProfile server, {
|
||
String host = '127.0.0.1',
|
||
int port = 6379,
|
||
String pass = '',
|
||
}) async {
|
||
await _connect(server);
|
||
final cli = _redisCli(host: host, port: port, pass: pass, db: 0);
|
||
final ks = await _capture('$cli INFO keyspace 2>&1', sudo: false);
|
||
final counts = <int, int>{};
|
||
for (final line in ks.split('\n')) {
|
||
final m = RegExp(r'^db(\d+):keys=(\d+)').firstMatch(line.trim());
|
||
if (m != null) counts[int.parse(m.group(1)!)] = int.parse(m.group(2)!);
|
||
}
|
||
final cfg = await _capture('$cli CONFIG GET databases 2>&1', sudo: false);
|
||
final nums = RegExp(
|
||
r'\d+',
|
||
).allMatches(cfg).map((m) => m.group(0)!).toList();
|
||
final dbCount = nums.isEmpty ? 16 : int.tryParse(nums.last) ?? 16;
|
||
return (dbCount, counts);
|
||
}
|
||
|
||
/// Scans keys returning (key, type) pairs, optionally filtered to
|
||
/// [filterType]. Uses a server-side Lua SCAN+TYPE (one round trip); scan is
|
||
/// capped at [limit] keys. Returns [] on error (caller falls back).
|
||
@override
|
||
Future<List<(String, String)>> redisKeysTyped(
|
||
ServerProfile server, {
|
||
String pattern = '*',
|
||
String filterType = '',
|
||
int limit = 50000,
|
||
String host = '127.0.0.1',
|
||
int port = 6379,
|
||
String pass = '',
|
||
int db = 0,
|
||
}) async {
|
||
await _connect(server);
|
||
final cli = _redisCli(host: host, port: port, pass: pass, db: db);
|
||
const script =
|
||
'local cur="0" local out={} local n=0 repeat local r=redis.call("SCAN",cur,"MATCH",ARGV[1],"COUNT",500) cur=r[1] for i=1,#r[2] do local k=r[2][i] local t=redis.call("TYPE",k).ok if ARGV[3]=="" or t==ARGV[3] then out[#out+1]=k out[#out+1]=t end n=n+1 if n>=tonumber(ARGV[2]) then cur="0" break end end until cur=="0" return out';
|
||
final out = await _capture(
|
||
"$cli EVAL '$script' 0 ${_shQuote(pattern)} ${_shQuote(limit.toString())} ${_shQuote(filterType)} 2>&1",
|
||
sudo: false,
|
||
);
|
||
final lines = out
|
||
.split('\n')
|
||
.map((l) => l.trim())
|
||
.where((l) => l.isNotEmpty)
|
||
.toList();
|
||
if (lines.isEmpty ||
|
||
out.contains('(error)') ||
|
||
out.toUpperCase().contains('ERR ') ||
|
||
out.contains('NOAUTH')) {
|
||
return [];
|
||
}
|
||
final result = <(String, String)>[];
|
||
for (var i = 0; i + 1 < lines.length; i += 2) {
|
||
result.add((lines[i], lines[i + 1]));
|
||
}
|
||
return result;
|
||
}
|
||
|
||
/// Lists keys matching [pattern] (SCAN-based; capped at [limit] for safety).
|
||
@override
|
||
Future<List<String>> redisKeys(
|
||
ServerProfile server, {
|
||
String pattern = '*',
|
||
int limit = 100000,
|
||
String host = '127.0.0.1',
|
||
int port = 6379,
|
||
String pass = '',
|
||
int db = 0,
|
||
}) async {
|
||
await _connect(server);
|
||
final cli = _redisCli(host: host, port: port, pass: pass, db: db);
|
||
final out = await _capture(
|
||
'$cli --scan --pattern ${_shQuote(pattern)} 2>/dev/null | head -n $limit',
|
||
sudo: false,
|
||
);
|
||
return out
|
||
.split('\n')
|
||
.map((l) => l.trim())
|
||
.where((l) => l.isNotEmpty)
|
||
.toList();
|
||
}
|
||
|
||
/// Reads a key's type, TTL and value (formatted per type).
|
||
@override
|
||
Future<RedisEntry> redisGet(
|
||
ServerProfile server,
|
||
String key, {
|
||
String host = '127.0.0.1',
|
||
int port = 6379,
|
||
String pass = '',
|
||
int db = 0,
|
||
}) async {
|
||
await _connect(server);
|
||
final cli = _redisCli(host: host, port: port, pass: pass, db: db);
|
||
final k = _shQuote(key);
|
||
final type = (await _capture('$cli TYPE $k 2>&1', sudo: false)).trim();
|
||
final ttl = (await _capture('$cli TTL $k 2>&1', sudo: false)).trim();
|
||
final readCmd = switch (type) {
|
||
'list' => '$cli LRANGE $k 0 -1',
|
||
'set' => '$cli SMEMBERS $k',
|
||
'hash' => '$cli HGETALL $k',
|
||
'zset' => '$cli ZRANGE $k 0 -1 WITHSCORES',
|
||
'stream' => '$cli XRANGE $k - +',
|
||
'string' => '$cli GET $k',
|
||
_ => '$cli GET $k',
|
||
};
|
||
final value = (await _capture('$readCmd 2>&1', sudo: false)).trim();
|
||
return RedisEntry(key: key, type: type, ttl: ttl, value: value);
|
||
}
|
||
|
||
// ---- docker ---------------------------------------------------------------
|
||
|
||
@override
|
||
Future<DockerStatus> dockerStatus(ServerProfile server) async {
|
||
await _connect(server);
|
||
final has = await _capture(
|
||
'command -v docker >/dev/null 2>&1 && echo yes || echo no',
|
||
sudo: false,
|
||
);
|
||
if (has.contains('no')) return DockerStatus(false, false, '');
|
||
final ver = (await _capture(
|
||
'docker --version 2>/dev/null',
|
||
sudo: false,
|
||
)).trim();
|
||
final info = await _capture(
|
||
'docker info >/dev/null 2>&1 && echo up || echo down',
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
);
|
||
return DockerStatus(true, info.contains('up'), ver);
|
||
}
|
||
|
||
static const _dockerSep = '~|~';
|
||
|
||
@override
|
||
Future<List<DockerContainer>> dockerPs(
|
||
ServerProfile server, {
|
||
bool all = true,
|
||
}) async {
|
||
await _connect(server);
|
||
final fmt =
|
||
'{{.ID}}$_dockerSep{{.Names}}$_dockerSep{{.Image}}$_dockerSep{{.State}}$_dockerSep{{.Status}}$_dockerSep{{.Ports}}';
|
||
final out = await _capture(
|
||
"docker ps ${all ? '-a' : ''} --format ${_shQuote(fmt)} 2>&1",
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
);
|
||
final list = <DockerContainer>[];
|
||
for (final line in out.split('\n')) {
|
||
final l = line.trim();
|
||
if (l.isEmpty) continue;
|
||
final parts = l.split(_dockerSep);
|
||
if (parts.length < 6) continue; // skip errors/noise
|
||
list.add(
|
||
DockerContainer(
|
||
id: parts[0],
|
||
name: parts[1],
|
||
image: parts[2],
|
||
state: parts[3],
|
||
status: parts[4],
|
||
ports: parts[5],
|
||
),
|
||
);
|
||
}
|
||
return list;
|
||
}
|
||
|
||
/// action: start | stop | restart | rm
|
||
@override
|
||
Future<String> dockerAction(
|
||
ServerProfile server,
|
||
String id,
|
||
String action,
|
||
) async {
|
||
await _connect(server);
|
||
final cmd = action == 'rm'
|
||
? 'docker rm -f ${_shQuote(id)}'
|
||
: 'docker $action ${_shQuote(id)}';
|
||
return (await _capture(
|
||
'$cmd 2>&1',
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
)).trim();
|
||
}
|
||
|
||
@override
|
||
Future<String> dockerLogs(
|
||
ServerProfile server,
|
||
String id, {
|
||
int tail = 300,
|
||
}) async {
|
||
await _connect(server);
|
||
return (await _capture(
|
||
'docker logs --tail $tail ${_shQuote(id)} 2>&1',
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
)).trim();
|
||
}
|
||
|
||
@override
|
||
Future<String> dockerTest(ServerProfile server) async {
|
||
await _connect(server);
|
||
return (await _capture(
|
||
'docker run --rm hello-world 2>&1',
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
)).trim();
|
||
}
|
||
|
||
/// Builds and runs `docker run`. Returns (ok, output).
|
||
@override
|
||
Future<(bool, String)> dockerRun(
|
||
ServerProfile server,
|
||
DockerRunSpec spec,
|
||
) async {
|
||
await _connect(server);
|
||
final b = StringBuffer('docker run');
|
||
b.write(spec.detached ? ' -d' : ' --rm');
|
||
if (spec.name.trim().isNotEmpty) {
|
||
b.write(' --name ${_shQuote(spec.name.trim())}');
|
||
}
|
||
if (spec.restart.trim().isNotEmpty && spec.detached) {
|
||
b.write(' --restart ${_shQuote(spec.restart.trim())}');
|
||
}
|
||
for (final p in spec.portList) {
|
||
b.write(' -p ${_shQuote(p)}');
|
||
}
|
||
for (final e in spec.envList) {
|
||
b.write(' -e ${_shQuote(e)}');
|
||
}
|
||
for (final v in spec.volumeList) {
|
||
b.write(' -v ${_shQuote(v)}');
|
||
}
|
||
b.write(' ${_shQuote(spec.image.trim())}');
|
||
if (spec.command.trim().isNotEmpty) b.write(' ${spec.command.trim()}');
|
||
|
||
_log(LogLevel.step, '\$ ${b.toString()}');
|
||
final out = await _capture(
|
||
'${b.toString()} 2>&1',
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
);
|
||
final t = out.trim();
|
||
// Success: docker prints the (64-hex) container id; errors contain text.
|
||
final ok =
|
||
RegExp(
|
||
r'^[0-9a-f]{12,64}$',
|
||
multiLine: true,
|
||
).hasMatch(t.split('\n').last.trim()) &&
|
||
!t.toLowerCase().contains('error');
|
||
_log(ok ? LogLevel.success : LogLevel.error, t);
|
||
return (ok, t);
|
||
}
|
||
|
||
// ---- kafka monitor --------------------------------------------------------
|
||
|
||
/// Full path to a Kafka CLI script, single-quoted for the shell.
|
||
String _kbin(String binDir, String suffix, String name) =>
|
||
_shQuote('$binDir/$name$suffix');
|
||
|
||
List<int> _ints(String s) => s
|
||
.split(',')
|
||
.map((e) => e.trim())
|
||
.where((e) => e.isNotEmpty)
|
||
.map((e) => int.tryParse(e) ?? -1)
|
||
.where((e) => e >= 0)
|
||
.toList();
|
||
|
||
/// Locates the Kafka tooling (bin dir + script suffix) on the server.
|
||
@override
|
||
Future<KafkaStatus> kafkaDetect(ServerProfile server) async {
|
||
await _connect(server);
|
||
final java = await _capture(
|
||
'command -v java >/dev/null 2>&1 && echo yes || echo no',
|
||
sudo: false,
|
||
);
|
||
// Prefer PATH, then fall back to the usual install locations.
|
||
const finder =
|
||
'p=""; '
|
||
'for c in kafka-topics.sh kafka-topics; do '
|
||
'if command -v "\$c" >/dev/null 2>&1; then p=\$(command -v "\$c"); break; fi; done; '
|
||
'if [ -z "\$p" ]; then '
|
||
'for d in /opt/kafka/bin /usr/local/kafka/bin /opt/bitnami/kafka/bin '
|
||
'/home/kafka/kafka/bin /usr/share/kafka/bin /opt/confluent/bin '
|
||
'/usr/lib/kafka/bin \$(ls -d /opt/kafka*/bin 2>/dev/null) '
|
||
'\$(ls -d /opt/kafka_*/bin 2>/dev/null) \$(ls -d /usr/local/kafka*/bin 2>/dev/null); do '
|
||
'for s in kafka-topics.sh kafka-topics; do '
|
||
'if [ -x "\$d/\$s" ]; then p="\$d/\$s"; break 2; fi; done; done; fi; '
|
||
'echo "\$p"';
|
||
final lines = (await _capture(
|
||
finder,
|
||
sudo: false,
|
||
)).split('\n').map((l) => l.trim()).where((l) => l.isNotEmpty).toList();
|
||
final path = lines.isEmpty ? '' : lines.last;
|
||
if (path.isEmpty) {
|
||
return KafkaStatus(
|
||
detected: false,
|
||
java: java.contains('yes'),
|
||
detail:
|
||
'kafka-topics not found on PATH or in the common install directories '
|
||
'(/opt/kafka, /usr/local/kafka, /opt/bitnami/kafka …).',
|
||
);
|
||
}
|
||
final slash = path.lastIndexOf('/');
|
||
final binDir = slash >= 0 ? path.substring(0, slash) : '.';
|
||
final base = slash >= 0 ? path.substring(slash + 1) : path;
|
||
final suffix = base.endsWith('.sh') ? '.sh' : '';
|
||
final ver = (await _capture(
|
||
'timeout 15 ${_shQuote(path)} --version 2>/dev/null | head -n1',
|
||
sudo: false,
|
||
)).trim();
|
||
return KafkaStatus(
|
||
detected: true,
|
||
binDir: binDir,
|
||
suffix: suffix,
|
||
java: java.contains('yes'),
|
||
version: ver,
|
||
detail: 'Found $path',
|
||
);
|
||
}
|
||
|
||
/// Connectivity probe — returns the broker-api-versions output (or the error).
|
||
@override
|
||
Future<String> kafkaTest(
|
||
ServerProfile server, {
|
||
required String binDir,
|
||
required String suffix,
|
||
required String bootstrap,
|
||
}) async {
|
||
await _connect(server);
|
||
final out = await _capture(
|
||
'timeout 25 ${_kbin(binDir, suffix, 'kafka-broker-api-versions')} '
|
||
'--bootstrap-server ${_shQuote(bootstrap)} 2>&1 | head -n 30',
|
||
sudo: false,
|
||
);
|
||
return out.trim();
|
||
}
|
||
|
||
/// topic:partition -> offset, for `--time` -1 (latest) or -2 (earliest).
|
||
Future<Map<String, int>> _kafkaOffsets(
|
||
String binDir,
|
||
String suffix,
|
||
String bootstrap,
|
||
int time,
|
||
) async {
|
||
var out = await _capture(
|
||
'timeout 25 ${_kbin(binDir, suffix, 'kafka-get-offsets')} '
|
||
'--bootstrap-server ${_shQuote(bootstrap)} --time $time 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
if (out.trim().isEmpty) {
|
||
// Older Kafka: fall back to the GetOffsetShell tool.
|
||
out = await _capture(
|
||
'timeout 25 ${_kbin(binDir, suffix, 'kafka-run-class')} '
|
||
'kafka.tools.GetOffsetShell --broker-list ${_shQuote(bootstrap)} '
|
||
'--time $time 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
}
|
||
final map = <String, int>{};
|
||
for (final raw in out.split('\n')) {
|
||
final m = RegExp(r'^([^:]+):(\d+):(-?\d+)$').firstMatch(raw.trim());
|
||
if (m != null) {
|
||
map['${m.group(1)}:${m.group(2)}'] = int.tryParse(m.group(3)!) ?? 0;
|
||
}
|
||
}
|
||
return map;
|
||
}
|
||
|
||
/// Lists every topic with its partitions, replication and message offsets.
|
||
@override
|
||
Future<List<KafkaTopic>> kafkaTopics(
|
||
ServerProfile server, {
|
||
required String binDir,
|
||
required String suffix,
|
||
required String bootstrap,
|
||
bool includeInternal = false,
|
||
}) async {
|
||
await _connect(server);
|
||
final describe = await _capture(
|
||
'timeout 25 ${_kbin(binDir, suffix, 'kafka-topics')} '
|
||
'--bootstrap-server ${_shQuote(bootstrap)} --describe 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
|
||
final topics = <KafkaTopic>[];
|
||
List<KafkaPartition> curParts = [];
|
||
for (final raw in describe.split('\n')) {
|
||
final line = raw.trim();
|
||
if (line.isEmpty) continue;
|
||
if (line.contains('PartitionCount:')) {
|
||
final name = RegExp(r'Topic:\s*(\S+)').firstMatch(line)?.group(1) ?? '';
|
||
final rf =
|
||
int.tryParse(
|
||
RegExp(
|
||
r'ReplicationFactor:\s*(\d+)',
|
||
).firstMatch(line)?.group(1) ??
|
||
'',
|
||
) ??
|
||
0;
|
||
final cfg =
|
||
RegExp(r'Configs:\s*(.*)').firstMatch(line)?.group(1)?.trim() ?? '';
|
||
curParts = [];
|
||
topics.add(
|
||
KafkaTopic(
|
||
name: name,
|
||
replicationFactor: rf,
|
||
partitions: curParts,
|
||
configs: cfg,
|
||
),
|
||
);
|
||
} else if (line.contains('Partition:') && line.contains('Leader:')) {
|
||
final pid =
|
||
int.tryParse(
|
||
RegExp(r'Partition:\s*(\d+)').firstMatch(line)?.group(1) ?? '',
|
||
) ??
|
||
0;
|
||
final leader =
|
||
int.tryParse(
|
||
RegExp(r'Leader:\s*(-?\d+)').firstMatch(line)?.group(1) ?? '',
|
||
) ??
|
||
-1;
|
||
final replicas = _ints(
|
||
RegExp(r'Replicas:\s*([0-9,]*)').firstMatch(line)?.group(1) ?? '',
|
||
);
|
||
final isr = _ints(
|
||
RegExp(r'Isr:\s*([0-9,]*)').firstMatch(line)?.group(1) ?? '',
|
||
);
|
||
curParts.add(
|
||
KafkaPartition(id: pid, leader: leader, replicas: replicas, isr: isr),
|
||
);
|
||
}
|
||
}
|
||
|
||
// Overlay message offsets (best effort — topics still render without them).
|
||
final latest = await _kafkaOffsets(binDir, suffix, bootstrap, -1);
|
||
final earliest = await _kafkaOffsets(binDir, suffix, bootstrap, -2);
|
||
for (final t in topics) {
|
||
for (final p in t.partitions) {
|
||
p.latest = latest['${t.name}:${p.id}'] ?? p.latest;
|
||
p.earliest = earliest['${t.name}:${p.id}'] ?? p.earliest;
|
||
}
|
||
}
|
||
|
||
final result = includeInternal
|
||
? topics
|
||
: topics.where((t) => !t.internal).toList();
|
||
result.sort((a, b) => a.name.compareTo(b.name));
|
||
return result;
|
||
}
|
||
|
||
/// Splits a Kafka CLI table row on runs of 2+ spaces (columns are padded).
|
||
List<String> _cols(String line) => line
|
||
.split(RegExp(r'\s{2,}'))
|
||
.map((e) => e.trim())
|
||
.where((e) => e.isNotEmpty)
|
||
.toList();
|
||
|
||
/// Lists consumer groups with state, members and per-partition lag.
|
||
@override
|
||
Future<List<KafkaConsumerGroup>> kafkaGroups(
|
||
ServerProfile server, {
|
||
required String binDir,
|
||
required String suffix,
|
||
required String bootstrap,
|
||
}) async {
|
||
await _connect(server);
|
||
final cg = _kbin(binDir, suffix, 'kafka-consumer-groups');
|
||
final bs = _shQuote(bootstrap);
|
||
|
||
final groups = <String, KafkaConsumerGroup>{};
|
||
|
||
// 1) All group ids (so idle/empty groups still appear).
|
||
final listOut = await _capture(
|
||
'timeout 25 $cg --bootstrap-server $bs --list 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
for (final raw in listOut.split('\n')) {
|
||
final id = raw.trim();
|
||
if (id.isEmpty || id.contains(' ')) continue;
|
||
groups[id] = KafkaConsumerGroup(groupId: id);
|
||
}
|
||
|
||
// 2) State / members / coordinator per group.
|
||
final stateOut = await _capture(
|
||
'timeout 30 $cg --bootstrap-server $bs --describe --all-groups --state 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
for (final raw in stateOut.split('\n')) {
|
||
final line = raw.trim();
|
||
if (line.isEmpty || line.startsWith('GROUP')) continue;
|
||
final c = _cols(line);
|
||
if (c.length < 4) continue;
|
||
final members = int.tryParse(c.last);
|
||
if (members == null) continue; // not a data row
|
||
final state = c[c.length - 2];
|
||
final id = c[0];
|
||
final g = groups.putIfAbsent(id, () => KafkaConsumerGroup(groupId: id));
|
||
g.state = state;
|
||
g.members = members;
|
||
}
|
||
|
||
// 3) Offsets / lag per topic-partition.
|
||
final offOut = await _capture(
|
||
'timeout 40 $cg --bootstrap-server $bs --describe --all-groups --offsets 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
for (final raw in offOut.split('\n')) {
|
||
final line = raw.trim();
|
||
if (line.isEmpty || line.startsWith('GROUP')) continue;
|
||
final c = _cols(line);
|
||
if (c.length < 6) continue;
|
||
final partition = int.tryParse(c[2]);
|
||
if (partition == null) continue; // header/noise
|
||
int num(String s) => int.tryParse(s.trim()) ?? 0;
|
||
final g = groups.putIfAbsent(
|
||
c[0],
|
||
() => KafkaConsumerGroup(groupId: c[0]),
|
||
);
|
||
g.assignments.add(
|
||
KafkaGroupAssignment(
|
||
topic: c[1],
|
||
partition: partition,
|
||
currentOffset: num(c[3]),
|
||
logEndOffset: num(c[4]),
|
||
lag: num(c[5]),
|
||
consumerId: c.length > 6 ? c[6] : '',
|
||
host: c.length > 7 ? c[7] : '',
|
||
clientId: c.length > 8 ? c[8] : '',
|
||
),
|
||
);
|
||
}
|
||
|
||
final list = groups.values.toList()
|
||
..sort((a, b) => a.groupId.compareTo(b.groupId));
|
||
return list;
|
||
}
|
||
|
||
/// Cluster snapshot: brokers, cluster id, controller and health counts.
|
||
@override
|
||
Future<KafkaCluster> kafkaCluster(
|
||
ServerProfile server, {
|
||
required String binDir,
|
||
required String suffix,
|
||
required String bootstrap,
|
||
}) async {
|
||
await _connect(server);
|
||
final bs = _shQuote(bootstrap);
|
||
|
||
// Brokers via broker-api-versions (one broker per header line).
|
||
final apiOut = await _capture(
|
||
'timeout 25 ${_kbin(binDir, suffix, 'kafka-broker-api-versions')} '
|
||
'--bootstrap-server $bs 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
final brokers = <KafkaBroker>[];
|
||
for (final raw in apiOut.split('\n')) {
|
||
final m = RegExp(
|
||
r'^(\S+):(\d+)\s+\(id:\s*(\d+)\s+rack:\s*(\S+)\)',
|
||
).firstMatch(raw.trim());
|
||
if (m != null) {
|
||
final rack = m.group(4)!;
|
||
brokers.add(
|
||
KafkaBroker(
|
||
id: int.parse(m.group(3)!),
|
||
host: m.group(1)!,
|
||
port: int.parse(m.group(2)!),
|
||
rack: rack == 'null' ? '' : rack,
|
||
),
|
||
);
|
||
}
|
||
}
|
||
brokers.sort((a, b) => a.id.compareTo(b.id));
|
||
|
||
// Cluster id (best effort).
|
||
String clusterId = '';
|
||
final cid = await _capture(
|
||
'timeout 20 ${_kbin(binDir, suffix, 'kafka-cluster')} cluster-id '
|
||
'--bootstrap-server $bs 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
clusterId =
|
||
RegExp(r'Cluster ID:\s*(\S+)').firstMatch(cid)?.group(1)?.trim() ?? '';
|
||
|
||
// Controller / quorum leader (KRaft) — best effort.
|
||
int controllerId = -1;
|
||
final quorum = await _capture(
|
||
'timeout 20 ${_kbin(binDir, suffix, 'kafka-metadata-quorum')} '
|
||
'--bootstrap-server $bs describe --status 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
controllerId =
|
||
int.tryParse(
|
||
RegExp(r'LeaderId:\s*(\d+)').firstMatch(quorum)?.group(1) ?? '',
|
||
) ??
|
||
-1;
|
||
for (final b in brokers) {
|
||
if (b.id == controllerId) b.isController = true;
|
||
}
|
||
|
||
// Health: under-replicated & unavailable partitions.
|
||
List<String> parsePartitions(String out) {
|
||
final res = <String>[];
|
||
for (final raw in out.split('\n')) {
|
||
final line = raw.trim();
|
||
final t = RegExp(r'Topic:\s*(\S+)').firstMatch(line)?.group(1);
|
||
final p = RegExp(r'Partition:\s*(\d+)').firstMatch(line)?.group(1);
|
||
if (t != null && p != null) res.add('$t-$p');
|
||
}
|
||
return res;
|
||
}
|
||
|
||
final ur = await _capture(
|
||
'timeout 25 ${_kbin(binDir, suffix, 'kafka-topics')} '
|
||
'--bootstrap-server $bs --describe --under-replicated-partitions 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
final un = await _capture(
|
||
'timeout 25 ${_kbin(binDir, suffix, 'kafka-topics')} '
|
||
'--bootstrap-server $bs --describe --unavailable-partitions 2>/dev/null',
|
||
sudo: false,
|
||
);
|
||
|
||
return KafkaCluster(
|
||
brokers: brokers,
|
||
clusterId: clusterId,
|
||
controllerId: controllerId,
|
||
underReplicated: parsePartitions(ur),
|
||
unavailable: parsePartitions(un),
|
||
);
|
||
}
|
||
|
||
String _unitFor(ServiceConfig s) => s.isWeb
|
||
? (s.nginxService.trim().isEmpty ? 'nginx' : s.nginxService.trim())
|
||
: s.serviceName.trim();
|
||
|
||
static String _fmtDt(DateTime d) {
|
||
String two(int n) => n.toString().padLeft(2, '0');
|
||
return '${d.year}-${two(d.month)}-${two(d.day)} '
|
||
'${two(d.hour)}:${two(d.minute)}:${two(d.second)}';
|
||
}
|
||
|
||
/// Public one-shot log fetch (connects if needed). Optionally restricts to a
|
||
/// time window via journalctl `--since` / `--until` (when using the default
|
||
/// journalctl command).
|
||
@override
|
||
Future<void> fetchLogs(
|
||
ServerProfile server,
|
||
ServiceConfig service, {
|
||
DateTime? since,
|
||
DateTime? until,
|
||
}) async {
|
||
await _connect(server);
|
||
_log(LogLevel.step, 'Fetching logs for ${service.name} …');
|
||
final hasRange = since != null || until != null;
|
||
if (hasRange && service.logCommand.trim().isEmpty) {
|
||
final unit = _unitFor(service).isEmpty ? 'app' : _unitFor(service);
|
||
final b = StringBuffer(
|
||
'journalctl -u ${_shQuote(unit)} -n 2000 --no-pager',
|
||
);
|
||
if (since != null) b.write(' --since ${_shQuote(_fmtDt(since))}');
|
||
if (until != null) b.write(' --until ${_shQuote(_fmtDt(until))}');
|
||
await _run(
|
||
b.toString(),
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
echo: LogLevel.remote,
|
||
);
|
||
} else {
|
||
await _runLogCommand(server, service);
|
||
}
|
||
}
|
||
|
||
/// The distinct `root` directives from the live nginx config (`nginx -T`),
|
||
/// so the deploy screen can warn if the web root doesn't match what nginx
|
||
/// actually serves. Returns [] if it can't be read.
|
||
@override
|
||
Future<List<String>> nginxRoots(ServerProfile server) async {
|
||
try {
|
||
await _connect(server);
|
||
final out = await _capture(
|
||
"nginx -T 2>/dev/null | grep -E '^[[:space:]]*root[[:space:]]' | "
|
||
"sed -E 's/^[[:space:]]*root[[:space:]]+//; s/;.*//; s/[[:space:]]+\$//' | sort -u",
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
);
|
||
return out
|
||
.split('\n')
|
||
.map((l) => l.trim())
|
||
.where((l) => l.isNotEmpty)
|
||
.toList();
|
||
} catch (_) {
|
||
return [];
|
||
}
|
||
}
|
||
|
||
/// Current systemd state of the service's unit ('active', 'inactive',
|
||
/// 'failed', 'activating', 'unknown', or 'no-unit').
|
||
@override
|
||
Future<String> serviceStatus(
|
||
ServerProfile server,
|
||
ServiceConfig service,
|
||
) async {
|
||
await _connect(server);
|
||
final unit = _unitFor(service);
|
||
if (unit.isEmpty) return 'no-unit';
|
||
final out = await _capture(
|
||
'systemctl is-active ${_shQuote(unit)} || true',
|
||
sudo: server.useSudo,
|
||
password: server.password,
|
||
);
|
||
final s = out.trim().split('\n').last.trim();
|
||
return s.isEmpty ? 'unknown' : s;
|
||
}
|
||
|
||
/// start | stop | restart the service's unit (reload for nginx web apps).
|
||
@override
|
||
Future<DeployResult> serviceControl(
|
||
ServerProfile server,
|
||
ServiceConfig service,
|
||
String action,
|
||
) async {
|
||
final unit = _unitFor(service);
|
||
if (unit.isEmpty) {
|
||
return DeployResult(
|
||
false,
|
||
false,
|
||
'No systemd unit set for this service.',
|
||
);
|
||
}
|
||
final sudo = server.useSudo;
|
||
final pw = server.password;
|
||
// nginx should reload, not restart, on "restart".
|
||
final effective = (service.isWeb && action == 'restart')
|
||
? 'reload'
|
||
: action;
|
||
try {
|
||
await _connect(server);
|
||
_log(LogLevel.step, 'systemctl $effective $unit …');
|
||
await _run(
|
||
'systemctl $effective ${_shQuote(unit)}',
|
||
sudo: sudo,
|
||
password: pw,
|
||
);
|
||
await Future.delayed(const Duration(seconds: 2));
|
||
final st = await serviceStatus(server, service);
|
||
final ok = action == 'stop'
|
||
? (st.contains('inactive') ||
|
||
st.contains('dead') ||
|
||
st.contains('failed'))
|
||
: (st.contains('active') && !st.contains('inactive'));
|
||
_log(ok ? LogLevel.success : LogLevel.warn, '$unit is now "$st".');
|
||
await _runLogCommand(server, service);
|
||
return DeployResult(ok, false, '$unit is $st.');
|
||
} catch (e) {
|
||
_log(LogLevel.error, 'Control failed: $e');
|
||
return DeployResult(false, false, 'Error: $e');
|
||
}
|
||
}
|
||
|
||
@override
|
||
Future<void> followLogs(ServerProfile server, ServiceConfig service) async {
|
||
await _connect(server);
|
||
stopLogs();
|
||
final unit = service.isWeb
|
||
? (service.nginxService.isEmpty ? 'nginx' : service.nginxService)
|
||
: (service.serviceName.isEmpty ? 'app' : service.serviceName);
|
||
final base = 'journalctl -u ${_shQuote(unit)} -n 50 -f';
|
||
final useSudo = server.useSudo && !_rootUser;
|
||
final cmd = useSudo ? "sudo -S -p '' $base" : base;
|
||
_log(LogLevel.step, 'Live tailing logs (press stop to end) …');
|
||
final session = await _client!.execute(cmd);
|
||
_logSession = session;
|
||
if (useSudo) {
|
||
session.stdin.add(
|
||
Uint8List.fromList(utf8.encode('${server.password}\n')),
|
||
);
|
||
}
|
||
await session.stdin.close();
|
||
unawaited(
|
||
session.stdout
|
||
.cast<List<int>>()
|
||
.transform(utf8.decoder)
|
||
.forEach((d) => _log(LogLevel.remote, d.trimRight())),
|
||
);
|
||
unawaited(
|
||
session.stderr
|
||
.cast<List<int>>()
|
||
.transform(utf8.decoder)
|
||
.forEach((d) => _log(LogLevel.remote, d.trimRight())),
|
||
);
|
||
}
|
||
|
||
@override
|
||
void stopLogs() {
|
||
_logSession?.close();
|
||
_logSession = null;
|
||
}
|
||
|
||
// ---- generic SFTP file transfer ------------------------------------------
|
||
|
||
@override
|
||
Future<List<RemoteEntry>> listDir(ServerProfile server, String path) async {
|
||
await _connect(server);
|
||
final sftp = await _client!.sftp();
|
||
final names = await sftp.listdir(path.isEmpty ? '.' : path);
|
||
final entries = <RemoteEntry>[];
|
||
for (final n in names) {
|
||
if (n.filename == '.' || n.filename == '..') continue;
|
||
entries.add(
|
||
RemoteEntry(
|
||
n.filename,
|
||
n.attr.isDirectory,
|
||
n.attr.size ?? 0,
|
||
n.attr.modifyTime,
|
||
),
|
||
);
|
||
}
|
||
entries.sort((a, b) {
|
||
if (a.isDir != b.isDir) return a.isDir ? -1 : 1;
|
||
return a.name.toLowerCase().compareTo(b.name.toLowerCase());
|
||
});
|
||
return entries;
|
||
}
|
||
|
||
@override
|
||
Future<String> resolveHome(ServerProfile server) async {
|
||
await _connect(server);
|
||
final sftp = await _client!.sftp();
|
||
try {
|
||
return await sftp.absolute('.');
|
||
} catch (_) {
|
||
return '.';
|
||
}
|
||
}
|
||
|
||
@override
|
||
Future<void> uploadFile(
|
||
ServerProfile server,
|
||
Uint8List bytes,
|
||
String remotePath,
|
||
) async {
|
||
await _connect(server);
|
||
await _sftpUpload(bytes, remotePath);
|
||
}
|
||
|
||
@override
|
||
Future<Uint8List> downloadFile(
|
||
ServerProfile server,
|
||
String remotePath,
|
||
) async {
|
||
await _connect(server);
|
||
final sftp = await _client!.sftp();
|
||
final file = await sftp.open(remotePath);
|
||
final attrs = await file.stat();
|
||
final total = attrs.size ?? 0;
|
||
final builder = BytesBuilder(copy: false);
|
||
final meter = _SpeedMeter();
|
||
await for (final chunk in file.read()) {
|
||
builder.add(chunk);
|
||
if (meter.tick(builder.length)) {
|
||
_emitTransfer(builder.length, total, meter, uploading: false);
|
||
}
|
||
}
|
||
await file.close();
|
||
_emitTransfer(
|
||
total == 0 ? builder.length : total,
|
||
total == 0 ? builder.length : total,
|
||
meter,
|
||
uploading: false,
|
||
force: true,
|
||
);
|
||
return builder.toBytes();
|
||
}
|
||
|
||
void _emitTransfer(
|
||
int sent,
|
||
int total,
|
||
_SpeedMeter meter, {
|
||
required bool uploading,
|
||
bool force = false,
|
||
}) {
|
||
final tp = TransferProgress(
|
||
sent,
|
||
total,
|
||
meter.bps,
|
||
meter.eta(sent, total),
|
||
uploading,
|
||
);
|
||
_progressController.add(tp);
|
||
if (_uploadStepKey != null) {
|
||
steps.progress(_uploadStepKey!, tp.fraction);
|
||
}
|
||
}
|
||
|
||
@override
|
||
Future<void> deleteRemote(
|
||
ServerProfile server,
|
||
String remotePath,
|
||
bool isDir,
|
||
) async {
|
||
await _connect(server);
|
||
final sftp = await _client!.sftp();
|
||
if (isDir) {
|
||
await sftp.rmdir(remotePath);
|
||
} else {
|
||
await sftp.remove(remotePath);
|
||
}
|
||
}
|
||
|
||
Future<void> _sftpUpload(Uint8List bytes, String remotePath) async {
|
||
final sftp = await _client!.sftp();
|
||
final file = await sftp.open(
|
||
remotePath,
|
||
mode:
|
||
SftpFileOpenMode.create |
|
||
SftpFileOpenMode.write |
|
||
SftpFileOpenMode.truncate,
|
||
);
|
||
final total = bytes.length;
|
||
var sent = 0;
|
||
const chunk = 32 * 1024;
|
||
// Chunks in flight per batch (~1 MB) — keeps the pipe full without
|
||
// dartssh2's SftpFileWriter, which leaks write errors as unhandled
|
||
// exceptions and never completes its future when the server rejects data.
|
||
const window = 32;
|
||
final meter = _SpeedMeter();
|
||
try {
|
||
for (var i = 0; i < total; i += chunk * window) {
|
||
final batch = <Future<void>>[];
|
||
for (var j = i; j < total && j < i + chunk * window; j += chunk) {
|
||
final end = j + chunk > total ? total : j + chunk;
|
||
batch.add(
|
||
file.writeBytes(Uint8List.sublistView(bytes, j, end), offset: j),
|
||
);
|
||
}
|
||
await Future.wait(batch);
|
||
sent = (i + chunk * window) > total ? total : i + chunk * window;
|
||
if (meter.tick(sent)) {
|
||
_emitTransfer(sent, total, meter, uploading: true);
|
||
}
|
||
}
|
||
} on SftpStatusError catch (e) {
|
||
try {
|
||
await file.close();
|
||
} catch (_) {}
|
||
try {
|
||
await sftp.remove(remotePath);
|
||
} catch (_) {}
|
||
throw Exception(
|
||
'SFTP write to $remotePath failed (${e.message}, code ${e.code})'
|
||
'${await _freeSpaceHint(sftp, remotePath, total)}',
|
||
);
|
||
}
|
||
await file.close();
|
||
_emitTransfer(total, total, meter, uploading: true, force: true);
|
||
}
|
||
|
||
/// Removes this run's /tmp staging files (no-op if they were already moved
|
||
/// into place) plus stale ones left by earlier crashed or cancelled runs.
|
||
/// Best-effort: never throws, so it can't mask the deploy's own result.
|
||
Future<void> _cleanStaging(ServerProfile s, List<String> paths) async {
|
||
try {
|
||
await _connect(s).timeout(const Duration(seconds: 20));
|
||
await _run(
|
||
'rm -rf ${paths.map(_shQuote).join(' ')}; '
|
||
"find /tmp -maxdepth 1 \\( -name 'peachdeploy-*' -o -name 'peachweb-*' \\) "
|
||
'-mmin +60 -exec rm -rf {} + 2>/dev/null; true',
|
||
sudo: s.useSudo,
|
||
password: s.password,
|
||
).timeout(const Duration(seconds: 30));
|
||
} catch (e) {
|
||
_log(LogLevel.warn, 'Could not clean staging files: $e');
|
||
}
|
||
}
|
||
|
||
/// Best-effort explanation for a generic SFTP write failure — usually the
|
||
/// target filesystem (often /tmp) is full or over quota.
|
||
Future<String> _freeSpaceHint(
|
||
SftpClient sftp,
|
||
String remotePath,
|
||
int needed,
|
||
) async {
|
||
try {
|
||
final slash = remotePath.lastIndexOf('/');
|
||
final dir = slash > 0 ? remotePath.substring(0, slash) : '/';
|
||
final vfs = await sftp.statvfs(dir);
|
||
if (vfs.isReadOnly) return ' — $dir is mounted read-only.';
|
||
final unit = vfs.fundamentalBlockSize > 0
|
||
? vfs.fundamentalBlockSize
|
||
: vfs.blockSize;
|
||
final free = unit * vfs.freeBlocksForNonRoot;
|
||
return ' — $dir has ${_fmtBytes(free)} free, upload needs '
|
||
'${_fmtBytes(needed)}. Free up space on the server.';
|
||
} catch (_) {
|
||
return ' — the remote filesystem rejected the data; check free space '
|
||
'and quota on the server (df -h).';
|
||
}
|
||
}
|
||
|
||
// ---- lifecycle ------------------------------------------------------------
|
||
|
||
@override
|
||
void disconnect() {
|
||
stopLogs();
|
||
_client?.close();
|
||
_client = null;
|
||
_connectedServerId = null;
|
||
_log(LogLevel.info, 'Disconnected.');
|
||
}
|
||
|
||
@override
|
||
void dispose() {
|
||
stopLogs();
|
||
_client?.close();
|
||
_client = null;
|
||
_connectedServerId = null;
|
||
_logController.close();
|
||
_progressController.close();
|
||
steps.dispose();
|
||
}
|
||
|
||
static String _fmtBytes(int b) {
|
||
if (b < 1024) return '$b B';
|
||
if (b < 1024 * 1024) return '${(b / 1024).toStringAsFixed(1)} KB';
|
||
return '${(b / (1024 * 1024)).toStringAsFixed(1)} MB';
|
||
}
|
||
}
|