aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/bin/einhyrningsinsctl.rs136
-rw-r--r--src/main.rs77
2 files changed, 209 insertions, 4 deletions
diff --git a/src/bin/einhyrningsinsctl.rs b/src/bin/einhyrningsinsctl.rs
new file mode 100644
index 0000000..c057b9e
--- /dev/null
+++ b/src/bin/einhyrningsinsctl.rs
@@ -0,0 +1,136 @@
+/*
+ * einhyrningsinsctl: controller/shell for einhyrningsins
+ * Copyright (C) 2016 Bryan Newbold <bnewbold@robocracy.org>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#[macro_use]
+extern crate json;
+
+extern crate getopts;
+extern crate log;
+extern crate env_logger;
+extern crate nix;
+extern crate timer;
+extern crate time;
+extern crate chan_signal;
+extern crate url;
+extern crate rustyline;
+
+
+use std::io::prelude::*;
+use std::io::{BufReader, BufWriter};
+use std::env;
+use std::path::Path;
+use std::process::exit;
+use std::os::unix::net::UnixStream;
+use getopts::Options;
+
+use rustyline::error::ReadlineError;
+use rustyline::Editor;
+
+
+// This is the main event loop
+fn shell(ctrl_stream: UnixStream) {
+
+ let mut reader = BufReader::new(&ctrl_stream);
+ let mut writer = BufWriter::new(&ctrl_stream);
+
+ // `()` can be used when no completer is required
+ let mut rl = Editor::<()>::new();
+
+ loop {
+ let readline = rl.readline("> ");
+ match readline {
+ Ok(line) => {
+ rl.add_history_entry(&line);
+ if line.len() == 0 { continue; };
+ let mut chunks = line.split(' ');
+ let cmd = chunks.nth(0).unwrap();
+ let args = chunks.collect();
+ send_msg(&mut reader, &mut writer, cmd, args).unwrap();
+ },
+ Err(ReadlineError::Interrupted) | Err(ReadlineError::Eof) => {
+ println!("Quitting...");
+ break
+ },
+ Err(err) => {
+ println!("Shell Error: {:?}", err);
+ break
+ }
+ }
+ }
+}
+
+fn send_msg(reader: &mut BufRead, writer: &mut Write, cmd: &str, args: Vec<&str>) -> Result<String, String> {
+ let mut buffer = String::new();
+ let mut arg_list = json::JsonValue::new_array();
+ for a in args {
+ arg_list.push(a).unwrap();
+ }
+ let req = object!{
+ "command" => cmd,
+ "args" => arg_list
+ };
+ //println!("Sending: {}", req.dump());
+ writer.write_all(req.dump().as_bytes()).unwrap();
+ writer.write_all("\n".as_bytes()).unwrap();
+ writer.flush().unwrap();
+
+ reader.read_line(&mut buffer).unwrap();
+ //println!("Got: {}", buffer);
+ let reply = json::parse(&buffer).unwrap();
+ println!("{}", reply.as_str().unwrap());
+ Ok(reply.as_str().unwrap().to_string())
+}
+
+fn print_usage(opts: Options) {
+ let brief = "usage:\teinhyrningsinsctl [options] program";
+ println!("");
+ print!("{}", opts.usage(&brief));
+}
+
+fn main() {
+
+ let args: Vec<String> = env::args().collect();
+
+ let mut opts = Options::new();
+ opts.optflag("h", "help", "print this help menu");
+
+ let matches = match opts.parse(&args[1..]) {
+ Ok(m) => { m }
+ Err(f) => { println!("{}", f.to_string()); print_usage(opts); exit(-1); }
+ };
+
+ if matches.opt_present("h") {
+ print_usage(opts);
+ return;
+ }
+
+ // Bind to Control Socket
+ let ctrl_path = Path::new("/tmp/einhorn.sock");
+ // XXX: handle this more gracefully (per-process)
+ if !ctrl_path.exists() {
+ println!("Couldn't find control socket: {:?}", ctrl_path);
+ exit(-1);
+ }
+ println!("Connecting to control socket: {:?}", ctrl_path);
+ let ctrl_stream = UnixStream::connect(ctrl_path).unwrap();
+
+ send_msg(&mut BufReader::new(&ctrl_stream), &mut BufWriter::new(&ctrl_stream), "ehlo", vec![]).unwrap();
+
+ shell(ctrl_stream);
+ exit(0);
+}
diff --git a/src/main.rs b/src/main.rs
index 0141a5f..725cdbd 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -18,6 +18,7 @@
#[macro_use]
extern crate chan;
+extern crate json;
extern crate getopts;
extern crate log;
@@ -47,7 +48,6 @@ use time::Duration;
use std::collections::HashMap;
use getopts::Options;
-use url::percent_encoding;
use chan_signal::Signal;
use chan::{Sender, Receiver};
use std::os::unix::io::{RawFd, IntoRawFd};
@@ -167,6 +167,7 @@ enum TimerAction {
enum CtrlAction {
Increment,
Decrement,
+ ManualAck(u32),
SigAll(Signal),
ShutdownAll,
UpgradeAll,
@@ -278,6 +279,14 @@ fn shepard(mut cfg: EinConfig, signal_rx: Receiver<Signal>) {
CtrlAction::Status => {
req.tx.send(format!("UNIMPLEMENTED"));
},
+ CtrlAction::ManualAck(pid) => {
+ if let Some(o) = brood.get_mut(&pid) {
+ if o.is_active() {
+ o.state = OffspringState::Healthy;
+ }
+ }
+ req.tx.send(format!("Acknowledged!"));
+ },
}
},
signal_rx.recv() -> sig => match sig.expect("Error with signal handler") {
@@ -508,14 +517,74 @@ fn ctrl_socket_handle(stream: UnixStream, ctrl_req_tx: Sender<CtrlRequest>) {
let reader = BufReader::new(&stream);
let mut writer = BufWriter::new(&stream);
for rawline in reader.lines() {
+
let rawline = rawline.unwrap();
- let line = percent_encoding::percent_decode(rawline.as_bytes()).decode_utf8().unwrap();
- println!("Decoded message: {}", line);
+ println!("Got line: {}", rawline);
+ if rawline.len() == 0 {
+ continue;
+ }
+
+ // Parse message
+ let req_action = if let Ok(msg) = json::parse(&rawline) {
+ match msg["command"].as_str() {
+ Some("worker:ack") => {
+ CtrlAction::ManualAck(msg["pid"].as_u32().unwrap())
+ },
+ Some("signal") => {
+ CtrlAction::SigAll(match msg["args"][0].as_str() {
+ Some("SIGHUP") | Some("HUP") => Signal::HUP,
+ Some("SIGINT") | Some("INT") => Signal::INT,
+ Some("SIGTERM") | Some("TERM") => Signal::TERM,
+ Some("SIGKILL") | Some("KILL") => Signal::KILL,
+ Some("SIGUSR1") | Some("USR1") => Signal::KILL,
+ Some("SIGUSR2") | Some("USR2") => Signal::USR2,
+ Some("SIGSTOP") | Some("STOP") => Signal::STOP,
+ Some("SIGCONT") | Some("CONT") => Signal::CONT,
+ Some(_) | None => {
+ writer.write_all("\"Missing or unhandled 'signal'\"\n".as_bytes()).unwrap();
+ writer.flush().unwrap();
+ continue;
+ },
+ })
+ },
+ Some("inc") => CtrlAction::Increment,
+ Some("dec") => CtrlAction::Decrement,
+ Some("status") => CtrlAction::Status,
+ Some("die") => CtrlAction::ShutdownAll,
+ Some("upgrade") => CtrlAction::UpgradeAll,
+ Some("ehlo") => {
+ writer.write_all("\"Hi there!\"\n\r".as_bytes()).unwrap();
+ writer.flush().unwrap();
+ continue;
+ },
+ Some("help") => {
+ writer.write_all("\"Command Listing: <TODO>\"\n".as_bytes()).unwrap(); // TODO
+ writer.flush().unwrap();
+ continue;
+ },
+ Some(_) | None => {
+ writer.write_all("\"Missing or unhandled 'command'\"\n".as_bytes()).unwrap();
+ writer.flush().unwrap();
+ continue;
+ },
+ }
+ } else {
+ writer.write_all("\"Expected valid JSON!\"\n".as_bytes()).unwrap();
+ writer.flush().unwrap();
+ continue;
+ };
+
+ // Send request
let (tx, rx): (Sender<String>, Receiver<String>) = chan::async();
- let req = CtrlRequest{ action: CtrlAction::Status, tx: tx };
+ let req = CtrlRequest{ action: req_action, tx: tx };
ctrl_req_tx.send(req);
+
+ // Send reply
let resp = rx.recv().unwrap();
+ writer.write_all("\"".as_bytes()).unwrap();
writer.write_all(resp.as_bytes()).unwrap();
+ writer.write_all("\"\n".as_bytes()).unwrap();
+ writer.flush().unwrap();
}
stream.shutdown(std::net::Shutdown::Both).unwrap();
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-21 22:24:06 +0000