From 2dca737392a208e094b6f054f39100b8d31ed80d Mon Sep 17 00:00:00 2001 From: bnewbold Date: Wed, 12 Oct 2016 00:22:29 -0700 Subject: partially working shell command --- src/bin/einhyrningsinsctl.rs | 136 +++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 77 ++++++++++++++++++++++-- 2 files changed, 209 insertions(+), 4 deletions(-) create mode 100644 src/bin/einhyrningsinsctl.rs (limited to 'src') 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 + * + * 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 . + */ + +#[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 { + 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 = 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) { 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) { 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: \"\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, Receiver) = 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(); } -- cgit v1.2.3