the-distro/ofborg
2
0
Fork 1
Code Issues 1 Pull requests 1 Packages Projects Releases Wiki Activity

Rustfmt

Browse source
This commit is contained in:
Graham Christensen 2018-01-21 14:17:25 -05:00
parent b1aa41b2de
commit e66776cee8
No known key found for this signature in database
GPG key ID: ACA1C1D120C83D5C
32 changed files with 927 additions and 758 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
ofborg/src/acl.rs
View file

@ -5,9 +5,7 @@ pub struct ACL {
impl ACL {
pub fn new(authorized_users: Vec<String>) -> ACL {
return ACL {
authorized_users: authorized_users,
}
return ACL { authorized_users: authorized_users };
}
pub fn can_build(&self, user: &str, repo: &str) -> bool {

10
ofborg/src/asynccmd.rs
View file

@ -40,9 +40,7 @@ fn reader_tx<R: 'static + Read + Send>(read: R, tx: Sender<String>) -> thread::J
impl AsyncCmd {
pub fn new(cmd: Command) -> AsyncCmd {
AsyncCmd {
command: cmd,
}
AsyncCmd { command: cmd }
}
pub fn spawn(mut self) -> SpawnedAsyncCmd {
@ -55,10 +53,8 @@ impl AsyncCmd {
let (tx, rx) = channel();
let stderr_handler = reader_tx(child.stderr.take().unwrap(),
tx.clone());
let stdout_handler = reader_tx(child.stdout.take().unwrap(),
tx.clone());
let stderr_handler = reader_tx(child.stderr.take().unwrap(), tx.clone());
let stdout_handler = reader_tx(child.stdout.take().unwrap(), tx.clone());
SpawnedAsyncCmd {
stdout_handler: stdout_handler,

11
ofborg/src/bin/builder.rs
View file

@ -51,14 +51,16 @@ fn main() {
}
&None => {
warn!("Please define feedback.full_logs in your configuration to true or false!");
warn!("feedback.full_logs when true will cause the full build log to be sent back to the server, and be viewable by everyone.");
warn!("feedback.full_logs when true will cause the full build log to be sent back");
warn!("to the server, and be viewable by everyone.");
warn!("I strongly encourage everybody turn this on!");
full_logs = false;
}
}
channel.basic_prefetch(1).unwrap();
channel.basic_consume(
channel
.basic_consume(
notifyworker::new(tasks::build::BuildWorker::new(
cloner,
nix,
@ -72,8 +74,9 @@ fn main() {
false,
false,
false,
Table::new()
).unwrap();
Table::new(),
)
.unwrap();
channel.start_consuming();

10
ofborg/src/bin/github-comment-filter.rs
View file

@ -39,10 +39,11 @@ fn main() {
let mut channel = session.open_channel(2).unwrap();
channel.basic_prefetch(1).unwrap();
channel.basic_consume(
channel
.basic_consume(
worker::new(tasks::githubcommentfilter::GitHubCommentWorker::new(
cfg.acl(),
cfg.github()
cfg.github(),
)),
"build-inputs",
format!("{}-github-comment-filter", cfg.whoami()).as_ref(),
@ -50,8 +51,9 @@ fn main() {
false,
false,
false,
Table::new()
).unwrap();
Table::new(),
)
.unwrap();
channel.start_consuming();

10
ofborg/src/bin/mass-rebuilder.rs
View file

@ -45,11 +45,12 @@ fn main() {
nix,
cfg.github(),
cfg.runner.identity.clone(),
events
events,
);
channel.basic_prefetch(1).unwrap();
channel.basic_consume(
channel
.basic_consume(
worker::new(mrw),
"mass-rebuild-check-jobs",
format!("{}-mass-rebuild-checker", cfg.whoami()).as_ref(),
@ -57,8 +58,9 @@ fn main() {
false,
false,
false,
Table::new()
).unwrap();
Table::new(),
)
.unwrap();
channel.start_consuming();

12
ofborg/src/bin/simple-build.rs
View file

@ -17,9 +17,15 @@ fn main() {
let nix = cfg.nix();
match nix.safely_build_attrs(&Path::new("./"), "./default.nix", vec![String::from("hello"),]) {
Ok(mut out) => { print!("{}", file_to_str(&mut out)); }
Err(mut out) => { print!("{}", file_to_str(&mut out)) }
match nix.safely_build_attrs(
&Path::new("./"),
"./default.nix",
vec![String::from("hello")],
) {
Ok(mut out) => {
print!("{}", file_to_str(&mut out));
}
Err(mut out) => print!("{}", file_to_str(&mut out)),
}
}

38
ofborg/src/checkout.rs
View file

@ -9,18 +9,16 @@ use std::ffi::OsString;
use std::process::Command;
pub struct CachedCloner {
root: PathBuf
root: PathBuf,
}
pub fn cached_cloner(path: &Path) -> CachedCloner {
return CachedCloner{
root: path.to_path_buf()
}
return CachedCloner { root: path.to_path_buf() };
}
pub struct CachedProject {
root: PathBuf,
clone_url: String
clone_url: String,
}
pub struct CachedProjectCo {
@ -43,8 +41,8 @@ impl CachedCloner {
return CachedProject {
root: new_root,
clone_url: clone_url
}
clone_url: clone_url,
};
}
}
@ -60,7 +58,7 @@ impl CachedProject {
id: id,
clone_url: self.clone_from().clone(),
local_reference: self.clone_to().clone(),
})
});
}
fn prefetch_cache(&self) -> Result<PathBuf, Error> {
@ -91,7 +89,7 @@ impl CachedProjectCo {
// let build_dir = self.build_dir();
return Ok(self.clone_to().to_str().unwrap().to_string())
return Ok(self.clone_to().to_str().unwrap().to_string());
}
pub fn fetch_pr(&self, pr_id: u64) -> Result<(), Error> {
@ -107,7 +105,7 @@ impl CachedProjectCo {
lock.unlock();
if result.success() {
return Ok(())
return Ok(());
} else {
return Err(Error::new(ErrorKind::Other, "Failed to fetch PR"));
}
@ -144,7 +142,7 @@ impl CachedProjectCo {
lock.unlock();
if result.success() {
return Ok(())
return Ok(());
} else {
return Err(Error::new(ErrorKind::Other, "Failed to merge"));
}
@ -153,19 +151,19 @@ impl CachedProjectCo {
impl clone::GitClonable for CachedProjectCo {
fn clone_from(&self) -> String {
return self.clone_url.clone()
return self.clone_url.clone();
}
fn clone_to(&self) -> PathBuf {
let mut clone_path = self.root.clone();
clone_path.push(&self.id);
return clone_path
return clone_path;
}
fn lock_path(&self) -> PathBuf {
let mut lock_path = self.root.clone();
lock_path.push(format!("{}.lock", self.id));
return lock_path
return lock_path;
}
fn extra_clone_args(&self) -> Vec<&OsStr> {
@ -174,30 +172,28 @@ impl clone::GitClonable for CachedProjectCo {
OsStr::new("--shared"),
OsStr::new("--reference-if-able"),
local_ref,
]
];
}
}
impl clone::GitClonable for CachedProject {
fn clone_from(&self) -> String {
return self.clone_url.clone()
return self.clone_url.clone();
}
fn clone_to(&self) -> PathBuf {
let mut clone_path = self.root.clone();
clone_path.push("clone");
return clone_path
return clone_path;
}
fn lock_path(&self) -> PathBuf {
let mut clone_path = self.root.clone();
clone_path.push("clone.lock");
return clone_path
return clone_path;
}
fn extra_clone_args(&self) -> Vec<&OsStr> {
return vec![
OsStr::new("--bare"),
]
return vec![OsStr::new("--bare")];
}
}

35
ofborg/src/clone.rs
View file

@ -6,7 +6,7 @@ use std::process::Command;
use std::ffi::OsStr;
pub struct Lock {
lock: Option<fs::File>
lock: Option<fs::File>,
}
impl Lock {
@ -27,24 +27,22 @@ pub trait GitClonable {
match fs::File::create(self.lock_path()) {
Err(e) => {
warn!("Failed to create lock file {:?}: {}",
self.lock_path(), e
);
warn!("Failed to create lock file {:?}: {}", self.lock_path(), e);
return Err(e);
}
Ok(lock) => {
match lock.lock_exclusive() {
Err(e) => {
warn!("Failed to get exclusive lock on file {:?}: {}",
self.lock_path(), e
warn!(
"Failed to get exclusive lock on file {:?}: {}",
self.lock_path(),
e
);
return Err(e);
}
Ok(_) => {
debug!("Got lock on {:?}", self.lock_path());
return Ok(Lock{
lock: Some(lock)
});
return Ok(Lock { lock: Some(lock) });
}
}
}
@ -56,14 +54,15 @@ pub trait GitClonable {
let mut lock = self.lock()?;
if self.clone_to().is_dir() {
debug!("Found dir at {:?}, initial clone is done",
self.clone_to());
return Ok(())
debug!("Found dir at {:?}, initial clone is done", self.clone_to());
return Ok(());
}
info!("Initial cloning of {} to {:?}",
info!(
"Initial cloning of {} to {:?}",
self.clone_from(),
self.clone_to());
self.clone_to()
);
let result = Command::new("git")
.arg("clone")
@ -75,7 +74,7 @@ pub trait GitClonable {
lock.unlock();
if result.success() {
return Ok(())
return Ok(());
} else {
return Err(Error::new(ErrorKind::Other, "Failed to clone"));
}
@ -94,7 +93,7 @@ pub trait GitClonable {
lock.unlock();
if result.success() {
return Ok(())
return Ok(());
} else {
return Err(Error::new(ErrorKind::Other, "Failed to fetch"));
}
@ -126,7 +125,7 @@ pub trait GitClonable {
lock.unlock();
return Ok(())
return Ok(());
}
fn checkout(&self, git_ref: &OsStr) -> Result<(), Error> {
@ -143,7 +142,7 @@ pub trait GitClonable {
lock.unlock();
if result.success() {
return Ok(())
return Ok(());
} else {
return Err(Error::new(ErrorKind::Other, "Failed to checkout"));
}

188
ofborg/src/commentparser.rs
View file

@ -3,7 +3,7 @@ pub fn parse(text: &str) -> Option<Vec<Instruction>> {
let instructions: Vec<Instruction> = text.lines()
.map(|s| match parse_line(s) {
Some(instructions) => instructions,
None => vec![]
None => vec![],
})
.fold(vec![], |mut collector, mut inst| {
collector.append(&mut inst);
@ -13,13 +13,12 @@ pub fn parse(text: &str) -> Option<Vec<Instruction>> {
if instructions.len() == 0 {
return None;
} else {
return Some(instructions)
return Some(instructions);
}
}
pub fn parse_line(text: &str) -> Option<Vec<Instruction>> {
let tokens: Vec<String> = text.split_whitespace()
.map(|s| s.to_owned()).collect();
let tokens: Vec<String> = text.split_whitespace().map(|s| s.to_owned()).collect();
if tokens.len() < 2 {
return None;
@ -38,23 +37,18 @@ pub fn parse_line(text: &str) -> Option<Vec<Instruction>> {
for command in commands {
let (left, right) = command.split_at(1);
match left[0].as_ref() {
"build" => {
instructions.push(Instruction::Build(Subset::Nixpkgs, right.to_vec()))
}
"build" => instructions.push(Instruction::Build(Subset::Nixpkgs, right.to_vec())),
"test" => {
instructions.push(
Instruction::Build(Subset::NixOS,
instructions.push(Instruction::Build(
Subset::NixOS,
right
.into_iter()
.map(|attr| format!("tests.{}.x86_64-linux", attr))
.collect()
)
);
.collect(),
));
}
"eval" => {
instructions.push(Instruction::Eval)
}
"eval" => instructions.push(Instruction::Eval),
_ => {}
}
}
@ -65,8 +59,7 @@ pub fn parse_line(text: &str) -> Option<Vec<Instruction>> {
#[derive(PartialEq, Debug)]
pub enum Instruction {
Build(Subset, Vec<String>),
Eval
Eval,
}
#[derive(Serialize, Deserialize, Debug, PartialEq)]
@ -89,8 +82,10 @@ mod tests {
fn valid_trailing_instruction() {
assert_eq!(
Some(vec![Instruction::Eval]),
parse("/cc @grahamc for ^^
@GrahamcOfBorg eval")
parse(
"/cc @grahamc for ^^
@GrahamcOfBorg eval",
)
);
}
@ -101,50 +96,62 @@ mod tests {
#[test]
fn eval_comment() {
assert_eq!(Some(vec![Instruction::Eval]),
parse("@grahamcofborg eval"));
assert_eq!(Some(vec![Instruction::Eval]), parse("@grahamcofborg eval"));
}
#[test]
fn eval_and_build_comment() {
assert_eq!(Some(vec![
assert_eq!(
Some(vec![
Instruction::Eval,
Instruction::Build(Subset::Nixpkgs, vec![
String::from("foo"),
])
Instruction::Build(
Subset::Nixpkgs,
vec![String::from("foo")]
),
]),
parse("@grahamcofborg eval @grahamcofborg build foo"));
parse("@grahamcofborg eval @grahamcofborg build foo")
);
}
#[test]
fn build_and_eval_and_build_comment() {
assert_eq!(Some(vec![
Instruction::Build(Subset::Nixpkgs, vec![
String::from("bar"),
]),
assert_eq!(
Some(vec![
Instruction::Build(
Subset::Nixpkgs,
vec![String::from("bar")]
),
Instruction::Eval,
Instruction::Build(Subset::Nixpkgs, vec![
String::from("foo"),
])
Instruction::Build(
Subset::Nixpkgs,
vec![String::from("foo")]
),
]),
parse("
parse(
"
@grahamcofborg build bar
@grahamcofborg eval
@grahamcofborg build foo"));
@grahamcofborg build foo",
)
);
}
#[test]
fn complex_comment_with_paragraphs() {
assert_eq!(Some(vec![
Instruction::Build(Subset::Nixpkgs, vec![
String::from("bar"),
]),
assert_eq!(
Some(vec![
Instruction::Build(
Subset::Nixpkgs,
vec![String::from("bar")]
),
Instruction::Eval,
Instruction::Build(Subset::Nixpkgs, vec![
String::from("foo"),
])
Instruction::Build(
Subset::Nixpkgs,
vec![String::from("foo")]
),
]),
parse("
parse(
"
I like where you're going with this PR, so let's try it out!
@grahamcofborg build bar
@ -154,70 +161,109 @@ I noticed though that the target branch was broken, which should be fixed. Let's
@grahamcofborg eval
Also, just in case, let's try foo
@grahamcofborg build foo"));
@grahamcofborg build foo",
)
);
}
#[test]
fn build_and_eval_comment() {
assert_eq!(Some(vec![
Instruction::Build(Subset::Nixpkgs, vec![
String::from("foo"),
]),
assert_eq!(
Some(vec![
Instruction::Build(
Subset::Nixpkgs,
vec![String::from("foo")]
),
Instruction::Eval,
]),
parse("@grahamcofborg build foo @grahamcofborg eval"));
parse("@grahamcofborg build foo @grahamcofborg eval")
);
}
#[test]
fn build_comment() {
assert_eq!(Some(vec![Instruction::Build(Subset::Nixpkgs, vec![
String::from("foo"),
String::from("bar")
])]),
parse("@GrahamCOfBorg build foo bar
assert_eq!(
Some(vec![
Instruction::Build(
Subset::Nixpkgs,
vec![String::from("foo"), String::from("bar")]
),
]),
parse(
"@GrahamCOfBorg build foo bar
baz"));
baz",
)
);
}
#[test]
fn test_comment() {
assert_eq!(Some(vec![Instruction::Build(Subset::NixOS, vec![
assert_eq!(
Some(vec![
Instruction::Build(
Subset::NixOS,
vec![
String::from("tests.foo.x86_64-linux"),
String::from("tests.bar.x86_64-linux"),
String::from("tests.baz.x86_64-linux")
])]),
parse("@GrahamCOfBorg test foo bar baz"));
String::from("tests.baz.x86_64-linux"),
]
),
]),
parse("@GrahamCOfBorg test foo bar baz")
);
}
#[test]
fn build_comment_newlines() {
assert_eq!(Some(vec![Instruction::Build(Subset::Nixpkgs, vec![
assert_eq!(
Some(vec![
Instruction::Build(
Subset::Nixpkgs,
vec![
String::from("foo"),
String::from("bar"),
String::from("baz")
])]),
parse("@GrahamCOfBorg build foo bar baz"));
String::from("baz"),
]
),
]),
parse("@GrahamCOfBorg build foo bar baz")
);
}
#[test]
fn build_comment_lower() {
assert_eq!(Some(vec![Instruction::Build(Subset::Nixpkgs, vec![
assert_eq!(
Some(vec![
Instruction::Build(
Subset::Nixpkgs,
vec![
String::from("foo"),
String::from("bar"),
String::from("baz")
])]),
parse("@grahamcofborg build foo bar baz"));
String::from("baz"),
]
),
]),
parse("@grahamcofborg build foo bar baz")
);
}
#[test]
fn build_comment_lower_package_case_retained() {
assert_eq!(Some(vec![Instruction::Build(Subset::Nixpkgs, vec![
assert_eq!(
Some(vec![
Instruction::Build(
Subset::Nixpkgs,
vec![
String::from("foo"),
String::from("bar"),
String::from("baz.Baz")
])]),
parse("@grahamcofborg build foo bar baz.Baz"));
String::from("baz.Baz"),
]
),
]),
parse("@grahamcofborg build foo bar baz.Baz")
);
}
}

18
ofborg/src/commitstatus.rs
View file

@ -12,7 +12,13 @@ pub struct CommitStatus<'a> {
}
impl<'a> CommitStatus<'a> {
pub fn new(api: hubcaps::statuses::Statuses<'a>, sha: String, context: String, description: String, url: Option<String>) -> CommitStatus<'a> {
pub fn new(
api: hubcaps::statuses::Statuses<'a>,
sha: String,
context: String,
description: String,
url: Option<String>,
) -> CommitStatus<'a> {
let mut stat = CommitStatus {
api: api,
sha: sha,
@ -23,7 +29,7 @@ impl <'a> CommitStatus<'a> {
stat.set_url(url);
return stat
return stat;
}
pub fn set_url(&mut self, url: Option<String>) {
@ -40,13 +46,15 @@ impl <'a> CommitStatus<'a> {
}
pub fn set(&self, state: hubcaps::statuses::State) {
self.api.create(
self.api
.create(
self.sha.as_ref(),
&hubcaps::statuses::StatusOptions::builder(state)
.context(self.context.clone())
.description(self.description.clone())
.target_url(self.url.clone())
.build()
).expect("Failed to mark final status on commit");
.build(),
)
.expect("Failed to mark final status on commit");
}
}

34
ofborg/src/config.rs
View file

@ -50,7 +50,7 @@ pub struct GithubConfig {
#[derive(Serialize, Deserialize, Debug)]
pub struct RunnerConfig {
pub identity: String,
pub authorized_users: Option<Vec<String>>
pub authorized_users: Option<Vec<String>>,
}
#[derive(Serialize, Deserialize, Debug)]
@ -64,38 +64,34 @@ impl Config {
}
pub fn acl(&self) -> acl::ACL {
return acl::ACL::new(
self.runner.authorized_users
.clone()
.expect("fetching config's runner.authorized_users")
);
return acl::ACL::new(self.runner.authorized_users.clone().expect(
"fetching config's runner.authorized_users",
));
}
pub fn github(&self) -> Github {
Github::new(
"github.com/grahamc/ofborg",
// tls configured hyper client
Client::with_connector(
HttpsConnector::new(
NativeTlsClient::new().unwrap()
)
),
Credentials::Token(self.github.clone().unwrap().token)
Client::with_connector(HttpsConnector::new(NativeTlsClient::new().unwrap())),
Credentials::Token(self.github.clone().unwrap().token),
)
}
pub fn nix(&self) -> Nix {
if self.nix.build_timeout_seconds < 1200 {
error!("Note: {} is way too low for build_timeout_seconds!",
error!(
"Note: {} is way too low for build_timeout_seconds!",
self.nix.build_timeout_seconds
);
error!("Please set build_timeout_seconds to at least 1200");
panic!();
}
return Nix::new(self.nix.system.clone(),
return Nix::new(
self.nix.system.clone(),
self.nix.remote.clone(),
self.nix.build_timeout_seconds
self.nix.build_timeout_seconds,
);
}
}
@ -103,9 +99,13 @@ impl Config {
impl RabbitMQConfig {
pub fn as_uri(&self) -> String {
return format!("{}://{}:{}@{}//",
return format!(
"{}://{}:{}@{}//",
if self.ssl { "amqps" } else { "amqp" },
self.username, self.password, self.host);
self.username,
self.password,
self.host
);
}
}

1
ofborg/src/evalchecker.rs
View file

@ -10,7 +10,6 @@ pub struct EvalChecker {
cmd: String,
args: Vec<String>,
nix: nix::Nix,
}
impl EvalChecker {

4
ofborg/src/ghevent/common.rs
View file

@ -24,6 +24,4 @@ pub struct Issue {
}
#[derive(Serialize, Deserialize, Debug)]
pub struct PullRequest {
}
pub struct PullRequest {}

6
ofborg/src/locks.rs
View file

@ -10,14 +10,12 @@ pub trait Lockable {
fn lock(&self) -> Result<Lock, Error> {
let lock = fs::File::create(self.lock_path())?;
lock.lock_exclusive()?;
return Ok(Lock{
lock: Some(lock)
})
return Ok(Lock { lock: Some(lock) });
}
}
pub struct Lock {
lock: Option<fs::File>
lock: Option<fs::File>,
}
impl Lock {

8
ofborg/src/message/buildlogmsg.rs
View file

@ -2,16 +2,16 @@ use ofborg::message::{Pr,Repo};
#[derive(Serialize, Deserialize, Debug)]
pub struct BuildLogMsg {
pub system: String,
pub identity: String,
pub attempt_id: String,
pub line_number: u64,
pub output: String,
pub identity: String,
pub system: String,
pub attempt_id: String,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct BuildLogStart {
pub identity: String,
pub system: String,
pub identity: String,
pub attempt_id: String,
}

2
ofborg/src/message/buildresult.rs
View file

@ -6,5 +6,5 @@ pub struct BuildResult {
pub pr: Pr,
pub system: String,
pub output: Vec<String>,
pub success: bool
pub success: bool,
}

12
ofborg/src/message/massrebuildjob.rs
View file

@ -13,20 +13,14 @@ pub struct MassRebuildJob {
pub pr: Pr,
}
pub struct Actions {
}
pub struct Actions {}
impl Actions {
pub fn skip(&mut self, _job: &MassRebuildJob) -> worker::Actions {
return vec![
worker::Action::Ack
];
return vec![worker::Action::Ack];
}
pub fn done(&mut self, _job: &MassRebuildJob) -> worker::Actions {
return vec![
worker::Action::Ack
];
return vec![worker::Action::Ack];
}
}

3
ofborg/src/message/plasticheartbeat.rs
View file

@ -4,8 +4,7 @@ extern crate env_logger;
use serde_json;
#[derive(Serialize, Deserialize, Debug)]
pub struct PlasticHeartbeat {
}
pub struct PlasticHeartbeat {}
pub fn from(data: &Vec<u8>) -> Result<PlasticHeartbeat, serde_json::error::Error> {
return serde_json::from_slice(&data);

131
ofborg/src/nix.rs
View file

@ -10,7 +10,7 @@ use std::io::SeekFrom;
pub struct Nix {
system: String,
remote: String,
build_timeout: u16
build_timeout: u16,
}
impl Nix {
@ -19,7 +19,7 @@ impl Nix {
system: system,
remote: remote,
build_timeout: build_timeout,
}
};
}
pub fn with_system(&self, system: String) -> Nix {
@ -30,13 +30,23 @@ impl Nix {
};
}
pub fn safely_build_attrs(&self, nixpkgs: &Path, file: &str, attrs: Vec<String>) -> Result<File,File> {
pub fn safely_build_attrs(
&self,
nixpkgs: &Path,
file: &str,
attrs: Vec<String>,
) -> Result<File, File> {
let cmd = self.safely_build_attrs_cmd(nixpkgs, file, attrs);
return self.run(cmd, true);
}
pub fn safely_build_attrs_cmd(&self, nixpkgs: &Path, file: &str, attrs: Vec<String>) -> Command {
pub fn safely_build_attrs_cmd(
&self,
nixpkgs: &Path,
file: &str,
attrs: Vec<String>,
) -> Command {
let mut attrargs: Vec<String> = Vec::with_capacity(3 + (attrs.len() * 2));
attrargs.push(file.to_owned());
attrargs.push(String::from("--no-out-link"));
@ -49,7 +59,13 @@ impl Nix {
return self.safe_command("nix-build", nixpkgs, attrargs);
}
pub fn safely(&self, cmd: &str, nixpkgs: &Path, args: Vec<String>, keep_stdout: bool) -> Result<File,File> {
pub fn safely(
&self,
cmd: &str,
nixpkgs: &Path,
args: Vec<String>,
keep_stdout: bool,
) -> Result<File, File> {
return self.run(self.safe_command(cmd, nixpkgs, args), keep_stdout);
}
@ -66,18 +82,19 @@ impl Nix {
stdout = Stdio::null();
}
let status = cmd
.stdout(Stdio::from(stdout))
let status = cmd.stdout(Stdio::from(stdout))
.stderr(Stdio::from(stderr))
.status()
.expect(format!("Running a program ...").as_ref());
reader.seek(SeekFrom::Start(0)).expect("Seeking to Start(0)");
reader.seek(SeekFrom::Start(0)).expect(
"Seeking to Start(0)",
);
if status.success() {
return Ok(reader)
return Ok(reader);
} else {
return Err(reader)
return Err(reader);
}
}
@ -94,13 +111,18 @@ impl Nix {
command.env("NIX_REMOTE", &self.remote);
command.args(&["--show-trace"]);
command.args(&["--option", "restrict-eval", "true"]);
command.args(&["--option", "build-timeout", &format!("{}", self.build_timeout)]);
command.args(
&[
"--option",
"build-timeout",
&format!("{}", self.build_timeout),
],
);
command.args(&["--argstr", "system", &self.system]);
command.args(args);
return command;
}
}
#[cfg(test)]
@ -153,23 +175,18 @@ mod tests {
let total_requirements = require.len();
let mut missed_requirements: usize = 0;
let requirements_held: Vec<Result<String, String>> =
require.into_iter()
let requirements_held: Vec<Result<String, String>> = require
.into_iter()
.map(|line| line.to_owned())
.map(|line|
if buildlog.contains(&line) {
.map(|line| if buildlog.contains(&line) {
Ok(line)
} else {
missed_requirements += 1;
Err(line)
}
)
})
.collect();
let mut prefixes: Vec<String> = vec![
"".to_owned(),
"".to_owned(),
];
let mut prefixes: Vec<String> = vec!["".to_owned(), "".to_owned()];
if !expectation_held {
prefixes.push(format!(
@ -178,16 +195,14 @@ mod tests {
));
prefixes.push("".to_owned());
} else {
prefixes.push(format!(
"The run was expected to {:?}, and did.",
expected
));
prefixes.push(format!("The run was expected to {:?}, and did.", expected));
prefixes.push("".to_owned());
}
let mut suffixes = vec![
"".to_owned(),
format!("{} out of {} required lines matched.",
format!(
"{} out of {} required lines matched.",
(total_requirements - missed_requirements),
total_requirements
),
@ -199,20 +214,15 @@ mod tests {
}
suffixes.push("".to_owned());
let output_blocks: Vec<Vec<String>> = vec![
prefixes,
vec![buildlog, "".to_owned()],
suffixes,
];
let output_blocks: Vec<Vec<String>> =
vec![prefixes, vec![buildlog, "".to_owned()], suffixes];
let output_blocks_strings: Vec<String> =
output_blocks
let output_blocks_strings: Vec<String> = output_blocks
.into_iter()
.map(|lines| lines.join("\n"))
.collect();
let output: String = output_blocks_strings
.join("\n");
let output: String = output_blocks_strings.join("\n");
if expectation_held && missed_requirements == 0 {
} else {
@ -233,15 +243,14 @@ mod tests {
let ret: Result<File, File> = nix.safely_build_attrs(
build_path().as_path(),
"default.nix",
vec![String::from("success")]
vec![String::from("success")],
);
assert_run(ret, Expect::Pass, vec![
"-success.drv",
"building path(s)",
"hi",
"-success"
]);
assert_run(
ret,
Expect::Pass,
vec!["-success.drv", "building path(s)", "hi", "-success"],
);
}
#[test]
@ -251,15 +260,19 @@ mod tests {
let ret: Result<File, File> = nix.safely_build_attrs(
build_path().as_path(),
"default.nix",
vec![String::from("failed")]
vec![String::from("failed")],
);
assert_run(ret, Expect::Fail, vec![
assert_run(
ret,
Expect::Fail,
vec![
"-failed.drv",
"building path(s)",
"hi",
"failed to produce output path"
]);
"failed to produce output path",
],
);
}
#[test]
@ -267,14 +280,18 @@ mod tests {
let ret: Result<File, File> = nix().safely_build_attrs(
build_path().as_path(),
"default.nix",
vec![String::from("sandbox-violation")]
vec![String::from("sandbox-violation")],
);
assert_run(ret, Expect::Fail, vec![
assert_run(
ret,
Expect::Fail,
vec![
"error: while evaluating the attribute",
"access to path",
"is forbidden in restricted mode"
]);
"is forbidden in restricted mode",
],
);
}
@ -284,13 +301,17 @@ mod tests {
"nix-instantiate",
passing_eval_path().as_path(),
vec![],
true
true,
);
assert_run(ret, Expect::Pass, vec![
assert_run(
ret,
Expect::Pass,
vec![
"the result might be removed by the garbage collector",
"-failed.drv",
"-success.drv"
]);
"-success.drv",
],
);
}
}

52
ofborg/src/notifyworker.rs
View file

@ -13,8 +13,12 @@ pub trait SimpleNotifyWorker {
fn consumer(&self, job: &Self::J, notifier: &mut NotificationReceiver);
fn msg_to_job(&self, method: &Deliver, headers: &BasicProperties,
body: &Vec<u8>) -> Result<Self::J, String>;
fn msg_to_job(
&self,
method: &Deliver,
headers: &BasicProperties,
body: &Vec<u8>,
) -> Result<Self::J, String>;
}
pub trait NotificationReceiver {
@ -22,14 +26,12 @@ pub trait NotificationReceiver {
}
pub struct DummyNotificationReceiver {
pub actions: Vec<Action>
pub actions: Vec<Action>,
}
impl DummyNotificationReceiver {
pub fn new() -> DummyNotificationReceiver {
DummyNotificationReceiver {
actions: vec![],
}
DummyNotificationReceiver { actions: vec![] }
}
}
@ -60,45 +62,43 @@ impl<'a> NotificationReceiver for ChannelNotificationReceiver<'a> {
self.channel.basic_ack(self.delivery_tag, false).unwrap();
}
Action::NackRequeue => {
self.channel.basic_nack(self.delivery_tag, false, true).unwrap();
self.channel
.basic_nack(self.delivery_tag, false, true)
.unwrap();
}
Action::NackDump => {
self.channel.basic_nack(self.delivery_tag, false, false).unwrap();
self.channel
.basic_nack(self.delivery_tag, false, false)
.unwrap();
}
Action::Publish(msg) => {
let exch = msg.exchange.clone().unwrap_or("".to_owned());
let key = msg.routing_key.clone().unwrap_or("".to_owned());
let props = msg.properties.unwrap_or(BasicProperties{ ..Default::default()});
self.channel.basic_publish(
exch,
key,
msg.mandatory,
msg.immediate,
props,
msg.content
).unwrap();
let props = msg.properties.unwrap_or(
BasicProperties { ..Default::default() },
);
self.channel
.basic_publish(exch, key, msg.mandatory, msg.immediate, props, msg.content)
.unwrap();
}
}
}
}
pub fn new<T: SimpleNotifyWorker>(worker: T) -> NotifyWorker<T> {
return NotifyWorker{
internal: worker,
};
return NotifyWorker { internal: worker };
}
impl<T: SimpleNotifyWorker + Send> Consumer for NotifyWorker<T> {
fn handle_delivery(&mut self,
fn handle_delivery(
&mut self,
channel: &mut Channel,
method: Deliver,
headers: BasicProperties,
body: Vec<u8>) {
let mut receiver = ChannelNotificationReceiver::new(
channel,
method.delivery_tag
);
body: Vec<u8>,
) {
let mut receiver = ChannelNotificationReceiver::new(channel, method.delivery_tag);
let job = self.internal.msg_to_job(&method, &headers, &body).unwrap();
self.internal.consumer(&job, &mut receiver);

16
ofborg/src/outpathdiff.rs
View file

@ -34,7 +34,7 @@ impl OutPathDiff {
.lines()
.filter_map(|line| match line {
Ok(line) => Some(line),
Err(_) => None
Err(_) => None,
})
.map(|x| {
let split: Vec<&str> = x.split_whitespace().collect();
@ -43,7 +43,8 @@ impl OutPathDiff {
} else {
info!("Warning: not 2 word segments in {:?}", split);
}
}).count();
})
.count();
}
return result;
@ -132,12 +133,13 @@ impl OutPaths {
self.place_nix();
let ret = self.execute();
self.remove_nix();
return ret
return ret;
}
fn place_nix(&self) {
let mut file = File::create(self.nix_path()).expect("Failed to create nix out path check");
file.write_all(include_str!("outpaths.nix").as_bytes()).expect("Failed to place outpaths.nix");
file.write_all(include_str!("outpaths.nix").as_bytes())
.expect("Failed to place outpaths.nix");
}
fn remove_nix(&self) {
@ -169,9 +171,11 @@ impl OutPaths {
String::from("-qaP"),
String::from("--no-name"),
String::from("--out-path"),
String::from("--arg"), String::from("checkMeta"), check_meta,
String::from("--arg"),
String::from("checkMeta"),
check_meta,
],
true
true,
)
}
}

14
ofborg/src/stats.rs
View file

@ -7,27 +7,27 @@ pub trait SysEvents {
}
pub struct RabbitMQ {
channel: Channel
channel: Channel,
}
impl RabbitMQ {
pub fn new(channel: Channel) -> RabbitMQ {
RabbitMQ {
channel: channel
}
RabbitMQ { channel: channel }
}
}
impl SysEvents for RabbitMQ {
fn tick(&mut self, name: &str) {
let props = BasicProperties { ..Default::default() };
self.channel.basic_publish(
self.channel
.basic_publish(
String::from("stats"),
"".to_owned(),
false,
false,
props,
String::from(name).into_bytes()
).unwrap();
String::from(name).into_bytes(),
)
.unwrap();
}
}

34
ofborg/src/tagger.rs
View file

@ -33,7 +33,10 @@ impl StdenvTagger {
for tag in &self.selected {
if !self.possible.contains(&tag) {
panic!("Tried to add label {} but it isn't in the possible list!", tag);
panic!(
"Tried to add label {} but it isn't in the possible list!",
tag
);
}
}
}
@ -88,23 +91,38 @@ impl RebuildTagger {
for attr in attrs {
match attr.rsplit(".").next() {
Some("x86_64-darwin") => { counter_darwin += 1; }
Some("x86_64-linux") => { counter_linux += 1; }
Some("x86_64-darwin") => {
counter_darwin += 1;
}
Some("x86_64-linux") => {
counter_linux += 1;
}
Some("aarch64-linux") => {}
Some("i686-linux") => {}
Some(arch) => { info!("Unknown arch: {:?}", arch); }
None => { info!("Cannot grok attr: {:?}", attr); }
Some(arch) => {
info!("Unknown arch: {:?}", arch);
}
None => {
info!("Cannot grok attr: {:?}", attr);
}
}
}
self.selected = vec![
self.selected =
vec![
String::from(format!("10.rebuild-linux: {}", self.bucket(counter_linux))),
String::from(format!("10.rebuild-darwin: {}", self.bucket(counter_darwin))),
String::from(format!(
"10.rebuild-darwin: {}",
self.bucket(counter_darwin)
)),
];
for tag in &self.selected {
if !self.possible.contains(&tag) {
panic!("Tried to add label {} but it isn't in the possible list!", tag);
panic!(
"Tried to add label {} but it isn't in the possible list!",
tag
);
}
}
}

96
ofborg/src/tasks/build.rs
View file

@ -28,7 +28,13 @@ pub struct BuildWorker {
}
impl BuildWorker {
pub fn new(cloner: checkout::CachedCloner, nix: nix::Nix, system: String, identity: String, full_logs: bool) -> BuildWorker {
pub fn new(
cloner: checkout::CachedCloner,
nix: nix::Nix,
system: String,
identity: String,
full_logs: bool,
) -> BuildWorker {
return BuildWorker {
cloner: cloner,
nix: nix,
@ -38,7 +44,11 @@ impl BuildWorker {
};
}
fn actions<'a, 'b>(&self, job: &'b buildjob::BuildJob, receiver: &'a mut notifyworker::NotificationReceiver) -> JobActions<'a, 'b> {
fn actions<'a, 'b>(
&self,
job: &'b buildjob::BuildJob,
receiver: &'a mut notifyworker::NotificationReceiver,
) -> JobActions<'a, 'b> {
JobActions::new(&self.system, &self.identity, job, receiver)
}
}
@ -55,8 +65,16 @@ struct JobActions<'a, 'b> {
}
impl<'a, 'b> JobActions<'a, 'b> {
fn new(system: &str, identity: &str, job: &'b buildjob::BuildJob, receiver: &'a mut notifyworker::NotificationReceiver) -> JobActions<'a, 'b> {
let (log_exchange, log_routing_key) = job.logs.clone().unwrap_or((String::from("logs"), String::from("build.log")));
fn new(
system: &str,
identity: &str,
job: &'b buildjob::BuildJob,
receiver: &'a mut notifyworker::NotificationReceiver,
) -> JobActions<'a, 'b> {
let (log_exchange, log_routing_key) = job.logs.clone().unwrap_or((
String::from("logs"),
String::from("build.log"),
));
return JobActions {
system: system.to_owned(),
@ -85,13 +103,13 @@ impl<'a, 'b> JobActions<'a, 'b> {
system: self.system.clone(),
output: vec![String::from("Merge failed")],
success: false
success: false,
};
self.tell(worker::publish_serde_action(
Some("build-results".to_owned()),
None,
&msg
&msg,
));
self.tell(worker::Action::Ack);
}
@ -111,7 +129,7 @@ impl<'a, 'b> JobActions<'a, 'b> {
self.tell(worker::publish_serde_action(
log_exchange,
log_routing_key,
&msg
&msg,
));
}
@ -132,7 +150,7 @@ impl<'a, 'b> JobActions<'a, 'b> {
self.tell(worker::publish_serde_action(
log_exchange,
log_routing_key,
&msg
&msg,
));
}
@ -142,13 +160,13 @@ impl<'a, 'b> JobActions<'a, 'b> {
pr: self.job.pr.clone(),
system: self.system.clone(),
output: lines,
success: success
success: success,
};
self.tell(worker::publish_serde_action(
Some("build-results".to_owned()),
None,
&msg
&msg,
));
self.tell(worker::Action::Ack);
}
@ -161,34 +179,46 @@ impl<'a, 'b> JobActions<'a, 'b> {
impl notifyworker::SimpleNotifyWorker for BuildWorker {
type J = buildjob::BuildJob;
fn msg_to_job(&self, _: &Deliver, _: &BasicProperties,
body: &Vec<u8>) -> Result<Self::J, String> {
fn msg_to_job(
&self,
_: &Deliver,
_: &BasicProperties,
body: &Vec<u8>,
) -> Result<Self::J, String> {
println!("lmao I got a job?");
return match buildjob::from(body) {
Ok(e) => { Ok(e) }
Ok(e) => Ok(e),
Err(e) => {
println!("{:?}", String::from_utf8(body.clone()));
panic!("{:?}", e);
}
}
};
}
fn consumer(&self, job: &buildjob::BuildJob, notifier: &mut notifyworker::NotificationReceiver) {
fn consumer(
&self,
job: &buildjob::BuildJob,
notifier: &mut notifyworker::NotificationReceiver,
) {
let mut actions = self.actions(&job, notifier);
info!("Working on {}", job.pr.number);
let project = self.cloner.project(job.repo.full_name.clone(), job.repo.clone_url.clone());
let co = project.clone_for("builder".to_string(),
self.identity.clone()).unwrap();
let project = self.cloner.project(
job.repo.full_name.clone(),
job.repo.clone_url.clone(),
);
let co = project
.clone_for("builder".to_string(), self.identity.clone())
.unwrap();
let target_branch = match job.pr.target_branch.clone() {
Some(x) => { x }
None => { String::from("origin/master") }
Some(x) => x,
None => String::from("origin/master"),
};
let buildfile = match job.subset {
Some(commentparser::Subset::NixOS) => "./nixos/release.nix",
_ => "./default.nix"
_ => "./default.nix",
};
// Note: Don't change the system limiter until the system isn't
@ -221,7 +251,7 @@ impl notifyworker::SimpleNotifyWorker for BuildWorker {
let cmd = self.nix.safely_build_attrs_cmd(
refpath.as_ref(),
buildfile,
job.attrs.clone()
job.attrs.clone(),
);
actions.log_started();
@ -254,10 +284,7 @@ impl notifyworker::SimpleNotifyWorker for BuildWorker {
let last10lines: Vec<String> = snippet_log.into_iter().collect::<Vec<String>>();
actions.build_finished(
success,
last10lines.clone()
);
actions.build_finished(success, last10lines.clone());
}
}
@ -294,7 +321,6 @@ mod tests {
fn make_worker() -> BuildWorker {
cleanup_scratch();
// pub fn new(cloner: checkout::CachedCloner, nix: nix::Nix, system: String, identity: String) -> BuildWorker {
let cloner = checkout::cached_cloner(&scratch_dir());
let nix = nix();
let worker = BuildWorker::new(
@ -302,7 +328,7 @@ mod tests {
nix,
"x86_64-linux".to_owned(),
"cargo-test-build".to_owned(),
true
true,
);
return worker;
@ -321,8 +347,8 @@ mod tests {
fn assert_contains_job(actions: &mut IntoIter<worker::Action>, text_to_match: &str) {
println!("\n\nSearching for {:?}", text_to_match);
actions.position(|job|
match job {
actions
.position(|job| match job {
worker::Action::Publish(ref body) => {
let mystr = String::from_utf8(body.content.clone()).unwrap();
if mystr.contains(text_to_match) {
@ -337,11 +363,11 @@ mod tests {
println!(" notPublish: {:?}", e);
return false;
}
}
).expect(
&format!("Actions should contain a job matching {:?}, after the previous check",
text_to_match)
);
})
.expect(&format!(
"Actions should contain a job matching {:?}, after the previous check",
text_to_match
));
}
#[test]

59
ofborg/src/tasks/githubcommentfilter.rs
View file

@ -14,7 +14,7 @@ use amqp::protocol::basic::{Deliver,BasicProperties};
pub struct GitHubCommentWorker {
acl: acl::ACL,
github: hubcaps::Github
github: hubcaps::Github,
}
impl GitHubCommentWorker {
@ -29,33 +29,42 @@ impl GitHubCommentWorker {
impl worker::SimpleWorker for GitHubCommentWorker {
type J = ghevent::IssueComment;
fn msg_to_job(&mut self, _: &Deliver, _: &BasicProperties,
body: &Vec<u8>) -> Result<Self::J, String> {
fn msg_to_job(
&mut self,
_: &Deliver,
_: &BasicProperties,
body: &Vec<u8>,
) -> Result<Self::J, String> {
return match serde_json::from_slice(body) {
Ok(e) => { Ok(e) }
Ok(e) => Ok(e),
Err(e) => {
println!("Failed to deserialize IsssueComment: {:?}", String::from_utf8(body.clone()));
println!(
"Failed to deserialize IsssueComment: {:?}",
String::from_utf8(body.clone())
);
panic!("{:?}", e);
}
}
};
}
fn consumer(&mut self, job: &ghevent::IssueComment) -> worker::Actions {
let instructions = commentparser::parse(&job.comment.body);
if instructions == None {
return vec![
worker::Action::Ack
];
return vec![worker::Action::Ack];
}
if !self.acl.can_build(&job.comment.user.login, &job.repository.full_name) {
println!("ACL prohibits {} from building {:?} for {}",
if !self.acl.can_build(
&job.comment.user.login,
&job.repository.full_name,
)
{
println!(
"ACL prohibits {} from building {:?} for {}",
job.comment.user.login,
instructions,
job.repository.full_name);
return vec![
worker::Action::Ack
];
job.repository.full_name
);
return vec![worker::Action::Ack];
}
println!("Got job: {:?}", job);
@ -64,20 +73,22 @@ impl worker::SimpleWorker for GitHubCommentWorker {
println!("Instructions: {:?}", instructions);
let pr = self.github
.repo(job.repository.owner.login.clone(), job.repository.name.clone())
.repo(
job.repository.owner.login.clone(),
job.repository.name.clone(),
)
.pulls()
.get(job.issue.number)
.get();
if let Err(x) = pr {
info!("fetching PR {}#{} from GitHub yielded error {}",
info!(
"fetching PR {}#{} from GitHub yielded error {}",
job.repository.full_name,
job.issue.number,
x
);
return vec![
worker::Action::Ack
];
return vec![worker::Action::Ack];
}
let pr = pr.unwrap();
@ -92,7 +103,7 @@ impl worker::SimpleWorker for GitHubCommentWorker {
let pr_msg = Pr {
number: job.issue.number.clone(),
head_sha: pr.head.sha.clone(),
target_branch: Some(pr.base.commit_ref.clone())
target_branch: Some(pr.base.commit_ref.clone()),
};
let mut response: Vec<worker::Action> = vec![];
@ -105,13 +116,13 @@ impl worker::SimpleWorker for GitHubCommentWorker {
pr: pr_msg.clone(),
subset: Some(subset),
attrs: attrs,
logs: Some(("logs".to_owned(), "build.log".to_owned()))
logs: Some(("logs".to_owned(), "build.log".to_owned())),
};
response.push(worker::publish_serde_action(
Some("build-jobs".to_owned()),
None,
&msg
&msg,
));
}
commentparser::Instruction::Eval => {
@ -123,7 +134,7 @@ impl worker::SimpleWorker for GitHubCommentWorker {
response.push(worker::publish_serde_action(
None,
Some("mass-rebuild-check-jobs".to_owned()),
&msg
&msg,
));
}

60
ofborg/src/tasks/heartbeat.rs
View file

@ -11,7 +11,7 @@ use std::process;
use amqp::Basic;
struct PlasticHeartbeatWorker {
queue_name: String
queue_name: String,
}
impl PlasticHeartbeatWorker {
@ -22,78 +22,80 @@ impl PlasticHeartbeatWorker {
mandatory: true,
immediate: false,
properties: None,
content: serde_json::to_string(&plasticheartbeat::PlasticHeartbeat{}).unwrap().into_bytes()
content: serde_json::to_string(&plasticheartbeat::PlasticHeartbeat {})
.unwrap()
.into_bytes(),
};
}
}
impl worker::SimpleWorker for PlasticHeartbeatWorker {
type J = plasticheartbeat::PlasticHeartbeat;
fn msg_to_job(&mut self, _: &Deliver, _: &BasicProperties,
body: &Vec<u8>) -> Result<Self::J, String> {
fn msg_to_job(
&mut self,
_: &Deliver,
_: &BasicProperties,
body: &Vec<u8>,
) -> Result<Self::J, String> {
return match plasticheartbeat::from(body) {
Ok(e) => { Ok(e) }
Ok(e) => Ok(e),
Err(e) => {
println!("{:?}", String::from_utf8(body.clone()));
panic!("{:?}", e);
}
}
};
}
fn consumer(&mut self, _job: &plasticheartbeat::PlasticHeartbeat) -> worker::Actions {
thread::sleep(time::Duration::from_secs(5));
return vec![
worker::Action::Publish(self.message()),
worker::Action::Ack
];
return vec![worker::Action::Publish(self.message()), worker::Action::Ack];
}
}
pub fn start_on_channel(mut hbchan: Channel, consumer_name: String) {
let queue_name = hbchan.queue_declare(
let queue_name = hbchan
.queue_declare(
"",
false, // passive
false, // durable
true, // exclusive
true, // auto_delete
false, //nowait
Table::new()
Table::new(),
)
.expect("Failed to declare an anon queue for PlasticHeartbeats!")
.queue;
println!("Got personal queue: {:?}", queue_name);
hbchan.basic_publish(
hbchan
.basic_publish(
"",
queue_name.as_ref(),
true, // mandatory
false, // immediate
BasicProperties {
..Default::default()
},
serde_json::to_string(&plasticheartbeat::PlasticHeartbeat{}).unwrap().into_bytes()
).unwrap();
BasicProperties { ..Default::default() },
serde_json::to_string(&plasticheartbeat::PlasticHeartbeat {})
.unwrap()
.into_bytes(),
)
.unwrap();
let worker = move ||
{
hbchan.basic_consume(
worker::new(
PlasticHeartbeatWorker{
queue_name: (&queue_name).clone()
}
),
let worker = move || {
hbchan
.basic_consume(
worker::new(PlasticHeartbeatWorker { queue_name: (&queue_name).clone() }),
queue_name,
String::from(format!("{}-heartbeat", consumer_name)),
false,
false,
false,
false,
Table::new()
).unwrap();
Table::new(),
)
.unwrap();
hbchan.start_consuming();
println!("PlasticHeartbeat failed");

225
ofborg/src/tasks/massrebuilder.rs
View file

@ -30,7 +30,13 @@ pub struct MassRebuildWorker<E> {
}
impl<E: stats::SysEvents> MassRebuildWorker<E> {
pub fn new(cloner: checkout::CachedCloner, nix: Nix, github: hubcaps::Github, identity: String, events: E) -> MassRebuildWorker<E> {
pub fn new(
cloner: checkout::CachedCloner,
nix: Nix,
github: hubcaps::Github,
identity: String,
events: E,
) -> MassRebuildWorker<E> {
return MassRebuildWorker {
cloner: cloner,
nix: nix,
@ -41,16 +47,19 @@ impl<E: stats::SysEvents> MassRebuildWorker<E> {
}
fn actions(&self) -> massrebuildjob::Actions {
return massrebuildjob::Actions{
};
return massrebuildjob::Actions {};
}
}
impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
type J = massrebuildjob::MassRebuildJob;
fn msg_to_job(&mut self, _: &Deliver, _: &BasicProperties,
body: &Vec<u8>) -> Result<Self::J, String> {
fn msg_to_job(
&mut self,
_: &Deliver,
_: &BasicProperties,
body: &Vec<u8>,
) -> Result<Self::J, String> {
self.events.tick("job-received");
return match massrebuildjob::from(body) {
Ok(e) => {
@ -59,15 +68,20 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
}
Err(e) => {
self.events.tick("job-decode-failure");
println!("{:?}", String::from_utf8(body.clone()));
panic!("{:?}", e);
}
println!(
"Failed to decode message: {:?}",
String::from_utf8(body.clone())
);
Err("Failed to decode message".to_owned())
}
};
}
fn consumer(&mut self, job: &massrebuildjob::MassRebuildJob) -> worker::Actions {
let repo = self.github
.repo(job.repo.owner.clone(), job.repo.name.clone());
let repo = self.github.repo(
job.repo.owner.clone(),
job.repo.name.clone(),
);
let gists = self.github.gists();
let issue = repo.issue(job.pr.number);
@ -92,48 +106,49 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
job.pr.head_sha.clone(),
"grahamcofborg-eval".to_owned(),
"Starting".to_owned(),
None
None,
);
overall_status.set_with_description("Starting", hubcaps::statuses::State::Pending);
let project = self.cloner.project(job.repo.full_name.clone(), job.repo.clone_url.clone());
let project = self.cloner.project(
job.repo.full_name.clone(),
job.repo.clone_url.clone(),
);
overall_status.set_with_description("Cloning project", hubcaps::statuses::State::Pending);
info!("Working on {}", job.pr.number);
let co = project.clone_for("mr-est".to_string(),
self.identity.clone()).unwrap();
let co = project
.clone_for("mr-est".to_string(), self.identity.clone())
.unwrap();
let target_branch = match job.pr.target_branch.clone() {
Some(x) => { x }
None => { String::from("master") }
Some(x) => x,
None => String::from("master"),
};
overall_status.set_with_description(
format!("Checking out {}", &target_branch).as_ref(),
hubcaps::statuses::State::Pending
hubcaps::statuses::State::Pending,
);
info!("Checking out target branch {}", &target_branch);
let refpath = co.checkout_origin_ref(target_branch.as_ref()).unwrap();
overall_status.set_with_description(
"Checking original stdenvs",
hubcaps::statuses::State::Pending
hubcaps::statuses::State::Pending,
);
let mut stdenvs = Stdenvs::new(self.nix.clone(), PathBuf::from(&refpath));
stdenvs.identify_before();
let mut rebuildsniff = OutPathDiff::new(
self.nix.clone(),
PathBuf::from(&refpath)
);
let mut rebuildsniff = OutPathDiff::new(self.nix.clone(), PathBuf::from(&refpath));
overall_status.set_with_description(
"Checking original out paths",
hubcaps::statuses::State::Pending
hubcaps::statuses::State::Pending,
);
if let Err(mut output) = rebuildsniff.find_before() {
@ -146,38 +161,32 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
overall_status.set_with_description(
format!("Target branch {} doesn't evaluate!", &target_branch).as_ref(),
hubcaps::statuses::State::Failure
hubcaps::statuses::State::Failure,
);
return self.actions().skip(&job);
}
overall_status.set_with_description(
"Fetching PR",
hubcaps::statuses::State::Pending
);
overall_status.set_with_description("Fetching PR", hubcaps::statuses::State::Pending);
co.fetch_pr(job.pr.number).unwrap();
if !co.commit_exists(job.pr.head_sha.as_ref()) {
overall_status.set_with_description(
"Commit not found",
hubcaps::statuses::State::Error
hubcaps::statuses::State::Error,
);
info!("Commit {} doesn't exist", job.pr.head_sha);
return self.actions().skip(&job);
}
overall_status.set_with_description(
"Merging PR",
hubcaps::statuses::State::Pending
);
overall_status.set_with_description("Merging PR", hubcaps::statuses::State::Pending);
if let Err(_) = co.merge_commit(job.pr.head_sha.as_ref()) {
overall_status.set_with_description(
"Failed to merge",
hubcaps::statuses::State::Failure
hubcaps::statuses::State::Failure,
);
info!("Failed to merge {}", job.pr.head_sha);
@ -186,14 +195,14 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
overall_status.set_with_description(
"Checking new stdenvs",
hubcaps::statuses::State::Pending
hubcaps::statuses::State::Pending,
);
stdenvs.identify_after();
overall_status.set_with_description(
"Checking new out paths",
hubcaps::statuses::State::Pending
hubcaps::statuses::State::Pending,
);
if let Err(mut output) = rebuildsniff.find_after() {
@ -204,8 +213,11 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
file_to_str(&mut output),
));
overall_status.set_with_description(
format!("Failed to enumerate outputs after merging to {}", &target_branch).as_ref(),
hubcaps::statuses::State::Failure
format!(
"Failed to enumerate outputs after merging to {}",
&target_branch
).as_ref(),
hubcaps::statuses::State::Failure,
);
return self.actions().skip(&job);
}
@ -213,11 +225,12 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
println!("Got path: {:?}, building", refpath);
overall_status.set_with_description(
"Beginning Evaluations",
hubcaps::statuses::State::Pending
hubcaps::statuses::State::Pending,
);
let eval_checks = vec![
EvalChecker::new("package-list",
EvalChecker::new(
"package-list",
"nix-env",
vec![
String::from("--file"),
@ -229,7 +242,8 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
self.nix.clone()
),
EvalChecker::new("nixos-options",
EvalChecker::new(
"nixos-options",
"nix-instantiate",
vec![
String::from("./nixos/release.nix"),
@ -239,7 +253,8 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
self.nix.clone()
),
EvalChecker::new("nixos-manual",
EvalChecker::new(
"nixos-manual",
"nix-instantiate",
vec![
String::from("./nixos/release.nix"),
@ -249,7 +264,8 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
self.nix.clone()
),
EvalChecker::new("nixpkgs-manual",
EvalChecker::new(
"nixpkgs-manual",
"nix-instantiate",
vec![
String::from("./pkgs/top-level/release.nix"),
@ -259,7 +275,8 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
self.nix.clone()
),
EvalChecker::new("nixpkgs-tarball",
EvalChecker::new(
"nixpkgs-tarball",
"nix-instantiate",
vec![
String::from("./pkgs/top-level/release.nix"),
@ -269,7 +286,8 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
self.nix.clone()
),
EvalChecker::new("nixpkgs-unstable-jobset",
EvalChecker::new(
"nixpkgs-unstable-jobset",
"nix-instantiate",
vec![
String::from("./pkgs/top-level/release.nix"),
@ -280,15 +298,15 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
),
];
let mut eval_results: bool = eval_checks.into_iter()
.map(|check|
{
let mut eval_results: bool = eval_checks
.into_iter()
.map(|check| {
let mut status = CommitStatus::new(
repo.statuses(),
job.pr.head_sha.clone(),
check.name(),
check.cli_cmd(),
None
None,
);
status.set(hubcaps::statuses::State::Pending);
@ -315,12 +333,11 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
status.set(state.clone());
if state == hubcaps::statuses::State::Success {
return Ok(())
return Ok(());
} else {
return Err(())
return Err(());
}
}
)
})
.all(|status| status == Ok(()));
if eval_results {
@ -329,7 +346,7 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
job.pr.head_sha.clone(),
String::from("grahamcofborg-eval-check-meta"),
String::from("config.nix: checkMeta = true"),
None
None,
);
status.set(hubcaps::statuses::State::Pending);
@ -337,11 +354,7 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
let state: hubcaps::statuses::State;
let gist_url: Option<String>;
let checker = OutPaths::new(
self.nix.clone(),
PathBuf::from(&refpath),
true
);
let checker = OutPaths::new(self.nix.clone(), PathBuf::from(&refpath), true);
match checker.find() {
Ok(_) => {
state = hubcaps::statuses::State::Success;
@ -366,32 +379,35 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
if eval_results {
overall_status.set_with_description(
"Calculating Changed Outputs",
hubcaps::statuses::State::Pending
hubcaps::statuses::State::Pending,
);
let mut stdenvtagger = StdenvTagger::new();
if !stdenvs.are_same() {
stdenvtagger.changed(stdenvs.changed());
}
update_labels(&issue, stdenvtagger.tags_to_add(),
stdenvtagger.tags_to_remove());
update_labels(
&issue,
stdenvtagger.tags_to_add(),
stdenvtagger.tags_to_remove(),
);
let mut rebuild_tags = RebuildTagger::new();
if let Some(attrs) = rebuildsniff.calculate_rebuild() {
rebuild_tags.parse_attrs(attrs);
}
update_labels(&issue, rebuild_tags.tags_to_add(),
rebuild_tags.tags_to_remove());
overall_status.set_with_description(
"^.^!",
hubcaps::statuses::State::Success
update_labels(
&issue,
rebuild_tags.tags_to_add(),
rebuild_tags.tags_to_remove(),
);
overall_status.set_with_description("^.^!", hubcaps::statuses::State::Success);
} else {
overall_status.set_with_description(
"Complete, with errors",
hubcaps::statuses::State::Failure
hubcaps::statuses::State::Failure,
);
}
@ -401,13 +417,13 @@ impl<E: stats::SysEvents> worker::SimpleWorker for MassRebuildWorker<E> {
enum StdenvFrom {
Before,
After
After,
}
#[derive(Debug)]
pub enum System {
X8664Darwin,
X8664Linux
X8664Linux,
}
#[derive(Debug, PartialEq)]
@ -433,7 +449,7 @@ impl Stdenvs {
darwin_stdenv_before: None,
darwin_stdenv_after: None,
}
};
}
fn identify_before(&mut self) {
@ -462,7 +478,7 @@ impl Stdenvs {
}
return changed
return changed;
}
fn identify(&mut self, system: System, from: StdenvFrom) {
@ -485,50 +501,62 @@ impl Stdenvs {
fn evalstdenv(&self, system: &str) -> Option<String> {
let result = self.nix.with_system(system.to_owned()).safely(
"nix-instantiate", &self.co, vec![
"nix-instantiate",
&self.co,
vec![
String::from("."),
String::from("-A"),
String::from("stdenv"),
],
true
true,
);
println!("{:?}", result);
return match result {
Ok(mut out) => {
file_to_drv(&mut out)
}
Ok(mut out) => file_to_drv(&mut out),
Err(mut out) => {
println!("{:?}", file_to_str(&mut out));
None
}
}
};
}
}
fn make_gist<'a>(gists: &hubcaps::gists::Gists<'a>, name: String, description: Option<String>, contents: String) -> Option<String> {
fn make_gist<'a>(
gists: &hubcaps::gists::Gists<'a>,
name: String,
description: Option<String>,
contents: String,
) -> Option<String> {
let mut files = HashMap::new();
files.insert(name.clone(),
files.insert(
name.clone(),
hubcaps::gists::Content {
filename: Some(name.clone()),
content: contents,
}
},
);
return Some(gists.create(
&hubcaps::gists::GistOptions {
return Some(
gists
.create(&hubcaps::gists::GistOptions {
description: description,
public: Some(true),
files: files,
}
).expect("Failed to create gist!").html_url);
})
.expect("Failed to create gist!")
.html_url,
);
}
pub fn update_labels(issue: &hubcaps::issues::IssueRef, add: Vec<String>, remove: Vec<String>) {
let l = issue.labels();
let existing: Vec<String> = issue.get().unwrap().labels
let existing: Vec<String> = issue
.get()
.unwrap()
.labels
.iter()
.map(|l| l.name.clone())
.collect();
@ -556,30 +584,31 @@ pub fn update_labels(issue: &hubcaps::issues::IssueRef, add: Vec<String>, remove
fn file_to_drv(f: &mut File) -> Option<String> {
let r = BufReader::new(f);
let matches: Vec<String>;
matches = r.lines().filter_map(|x|
match x {
matches = r.lines()
.filter_map(|x| match x {
Ok(line) => {
if !line.starts_with("/nix/store/") {
debug!("Skipping line, not /nix/store: {}", line);
return None
return None;
}
if !line.ends_with(".drv") {
debug!("Skipping line, not .drv: {}", line);
return None
return None;
}
return Some(line)
return Some(line);
}
Err(_) => None
}).collect();
Err(_) => None,
})
.collect();
if matches.len() == 1 {
return Some(matches.first().unwrap().clone());
} else {
info!("Got wrong number of matches: {}", matches.len());
info!("Matches: {:?}", matches);
return None
return None;
}
}
@ -597,7 +626,11 @@ mod tests {
#[test]
fn stdenv_checking() {
let nix = Nix::new(String::from("x86_64-linux"), String::from("daemon"), 1200);
let mut stdenv = Stdenvs::new(nix.clone(), PathBuf::from("/nix/var/nix/profiles/per-user/root/channels/nixos/nixpkgs"));
let mut stdenv =
Stdenvs::new(
nix.clone(),
PathBuf::from("/nix/var/nix/profiles/per-user/root/channels/nixos/nixpkgs"),
);
stdenv.identify(System::X8664Linux, StdenvFrom::Before);
stdenv.identify(System::X8664Darwin, StdenvFrom::Before);

57
ofborg/src/worker.rs
View file

@ -7,11 +7,10 @@ use serde_json;
use std::cmp::PartialEq;
pub struct Worker<T: SimpleWorker> {
internal: T
internal: T,
}
pub struct Response {
}
pub struct Response {}
pub type Actions = Vec<Action>;
@ -33,9 +32,14 @@ pub struct QueueMsg {
pub content: Vec<u8>,
}
pub fn publish_serde_action<T: ?Sized>(exchange: Option<String>, routing_key: Option<String>, msg: &T) -> Action
pub fn publish_serde_action<T: ?Sized>(
exchange: Option<String>,
routing_key: Option<String>,
msg: &T,
) -> Action
where
T: Serialize {
T: Serialize,
{
let props = BasicProperties {
content_type: Some("application/json".to_owned()),
..Default::default()
@ -47,7 +51,7 @@ pub fn publish_serde_action<T: ?Sized>(exchange: Option<String>, routing_key: Op
mandatory: true,
immediate: false,
properties: Some(props),
content: serde_json::to_string(&msg).unwrap().into_bytes()
content: serde_json::to_string(&msg).unwrap().into_bytes(),
});
}
@ -56,24 +60,28 @@ pub trait SimpleWorker {
fn consumer(&mut self, job: &Self::J) -> Actions;
fn msg_to_job(&mut self, method: &Deliver, headers: &BasicProperties,
body: &Vec<u8>) -> Result<Self::J, String>;
fn msg_to_job(
&mut self,
method: &Deliver,
headers: &BasicProperties,
body: &Vec<u8>,
) -> Result<Self::J, String>;
}
pub fn new<T: SimpleWorker>(worker: T) -> Worker<T> {
return Worker{
internal: worker,
};
return Worker { internal: worker };
}
impl<T: SimpleWorker + Send> Consumer for Worker<T> {
fn handle_delivery(&mut self,
fn handle_delivery(
&mut self,
channel: &mut Channel,
method: Deliver,
headers: BasicProperties,
body: Vec<u8>) {
body: Vec<u8>,
) {
@ -84,24 +92,25 @@ impl <T: SimpleWorker + Send> Consumer for Worker<T> {
channel.basic_ack(method.delivery_tag, false).unwrap();
}
Action::NackRequeue => {
channel.basic_nack(method.delivery_tag, false, true).unwrap();
channel
.basic_nack(method.delivery_tag, false, true)
.unwrap();
}
Action::NackDump => {
channel.basic_nack(method.delivery_tag, false, false).unwrap();
channel
.basic_nack(method.delivery_tag, false, false)
.unwrap();
}
Action::Publish(msg) => {
let exch = msg.exchange.clone().unwrap_or("".to_owned());
let key = msg.routing_key.clone().unwrap_or("".to_owned());
let props = msg.properties.unwrap_or(BasicProperties{ ..Default::default()});
channel.basic_publish(
exch,
key,
msg.mandatory,
msg.immediate,
props,
msg.content
).unwrap();
let props = msg.properties.unwrap_or(
BasicProperties { ..Default::default() },
);
channel
.basic_publish(exch, key, msg.mandatory, msg.immediate, props, msg.content)
.unwrap();
}
}
}

1
shell.nix
View file

@ -27,6 +27,7 @@ let
nix-prefetch-git
rust.rustc
rust.cargo
rustfmt
carnix
openssl.dev
pkgconfig