2222 lines
70 KiB
Dart
2222 lines
70 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'),
|
||
]);
|
||
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('/'));
|
||
final staging =
|
||
'/tmp/peachdeploy-${DateTime.now().millisecondsSinceEpoch}.jar';
|
||
|
||
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');
|
||
}
|
||
}
|
||
|
||
// ---- 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');
|
||
}
|
||
}
|
||
|
||
/// 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');
|
||
}
|
||
}
|
||
|
||
// ---- 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;
|
||
final meter = _SpeedMeter();
|
||
final stream =
|
||
Stream<Uint8List>.fromIterable([
|
||
for (var i = 0; i < total; i += chunk)
|
||
Uint8List.sublistView(
|
||
bytes,
|
||
i,
|
||
i + chunk > total ? total : i + chunk,
|
||
),
|
||
]).map((c) {
|
||
sent += c.length;
|
||
if (meter.tick(sent)) {
|
||
_emitTransfer(sent, total, meter, uploading: true);
|
||
}
|
||
return c;
|
||
});
|
||
await file.write(stream);
|
||
await file.close();
|
||
_emitTransfer(total, total, meter, uploading: true, force: true);
|
||
}
|
||
|
||
// ---- 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';
|
||
}
|
||
}
|