EX-1687 graphql commands (#2240)

* Method created

Signed-off-by: Nick Linker <nlinker@gmail.com>

* Debugging routine

Signed-off-by: Nick Linker <nlinker@gmail.com>

* Debugging wip

Signed-off-by: Nick Linker <nlinker@gmail.com>

* Finished

Signed-off-by: Nick Linker <nlinker@gmail.com>

* Fix more formatting

Signed-off-by: Nick Linker <nlinker@gmail.com>

* Using case insensitive search for a message

Signed-off-by: Nick Linker <nlinker@gmail.com>

* Regenerate sqlx to fix the CI error

Signed-off-by: Nick Linker <nlinker@gmail.com>

iml-api/src/graphql.rs

@@ -6,7 +6,7 @@ use crate::{command::get_command, error::ImlApiError};
 use futures::{TryFutureExt, TryStreamExt};
 use iml_postgres::{sqlx, PgPool};
 use iml_rabbit::Pool;
-use iml_wire_types::{snapshot::Snapshot, Command};
+use iml_wire_types::{snapshot::Snapshot, Command, EndpointName, Job};
 use itertools::Itertools;
 use juniper::{
  http::{graphiql::graphiql_source, GraphQLRequest},
@@ -284,10 +284,11 @@ impl QueryRoot {
 
  Ok(xs)
  }
+
  #[graphql(arguments(
  limit(description = "paging limit, defaults to 20"),
  offset(description = "Offset into items, defaults to 0"),
- dir(description = "Sort direction, defaults to asc"),
+ dir(description = "Sort direction, defaults to ASC"),
  fsname(description = "Filesystem the snapshot was taken from"),
  name(description = "Name of the snapshot"),
  ))]
@@ -322,6 +323,77 @@ impl QueryRoot {
 
  Ok(snapshots)
  }
+
+ /// Fetch the list of commands
+ #[graphql(arguments(
+ limit(description = "paging limit, defaults to 20"),
+ offset(description = "Offset into items, defaults to 0"),
+ dir(description = "Sort direction, defaults to ASC"),
+ is_active(description = "Command status, active means not completed, default is false"),
+ msg(description = "Substring of the command's message, null or empty matches all"),
+ ))]
+ async fn commands(
+ context: &Context,
+ limit: Option<i32>,
+ offset: Option<i32>,
+ dir: Option<SortDir>,
+ is_active: Option<bool>,
+ msg: Option<String>,
+ ) -> juniper::FieldResult<Vec<Command>> {
+ let dir = dir.unwrap_or_default();
+ let is_completed = !is_active.unwrap_or(false);
+ let commands: Vec<Command> = sqlx::query!(
+ r#"
+ SELECT
+ c.id AS id,
+ cancelled,
+ complete,
+ errored,
+ created_at,
+ array_agg(cj.job_id)::INT[] AS job_ids,
+ message
+ FROM chroma_core_command c
+ JOIN chroma_core_command_jobs cj ON c.id = cj.command_id
+ WHERE complete = $4
+ AND ($5::TEXT IS NULL OR c.message ILIKE '%' || $5 || '%')
+ GROUP BY c.id
+ ORDER BY
+ CASE WHEN $3 = 'asc' THEN c.id END ASC,
+ CASE WHEN $3 = 'desc' THEN c.id END DESC
+ OFFSET $1 LIMIT $2 "#,
+ offset.unwrap_or(0) as i64,
+ limit.unwrap_or(20) as i64,
+ dir.deref(),
+ is_completed,
+ msg,
+ )
+ .fetch_all(&context.pg_pool)
+ .map_ok(|xs: Vec<_>| {
+ xs.into_iter()
+ .map(|x| Command {
+ id: x.id,
+ cancelled: x.cancelled,
+ complete: x.complete,
+ errored: x.errored,
+ created_at: x.created_at.format("%Y-%m-%dT%T%.6f").to_string(),
+ jobs: {
+ x.job_ids
+ .unwrap_or_default()
+ .into_iter()
+ .map(|job_id: i32| {
+ format!("/api/{}/{}/", Job::<()>::endpoint_name(), job_id)
+ })
+ .collect::<Vec<_>>()
+ },
+ logs: "".to_string(),
+ message: x.message.clone(),
+ resource_uri: format!("/api/{}/{}/", Command::endpoint_name(), x.id),
+ })
+ .collect::<Vec<_>>()
+ })
+ .await?;
+ Ok(commands)
+ }
 }
 
 #[juniper::graphql_object(Context = Context)]

iml-manager-env/src/lib.rs

@@ -238,6 +238,10 @@ pub fn get_db_host() -> Option<String> {
  empty_str_to_none(get_var("DB_HOST"))
 }
 
+pub fn get_db_port() -> Option<u16> {
+ env::var("DB_PORT").ok().map(|l| l.parse().ok()).flatten()
+}
+
 pub fn get_db_name() -> Option<String> {
  empty_str_to_none(get_var("DB_NAME"))
 }

iml-postgres/src/lib.rs

@@ -30,6 +30,12 @@ pub async fn get_db_pool(pool_size: u32) -> Result<PgPool, sqlx::Error> {
  opts
  };
 
+ opts = if let Some(x) = iml_manager_env::get_db_port() {
+ opts.port(x)
+ } else {
+ opts
+ };
+
  opts = if let Some(x) = iml_manager_env::get_db_name() {
  opts.database(&x)
  } else {

sqlx-data.json

@@ -2034,6 +2034,66 @@
  ]
  }
  },
+ "96578c531d3344f51586bd7f502d258fe105acafebd361a5458d14fa4c26af8d": {
+ "query": "\n SELECT\n c.id AS id,\n cancelled,\n complete,\n errored,\n created_at,\n array_agg(cj.job_id)::INT[] AS job_ids,\n message\n FROM chroma_core_command c\n JOIN chroma_core_command_jobs cj ON c.id = cj.command_id\n WHERE complete = $4\n AND ($5::TEXT IS NULL OR c.message ILIKE '%' || $5 || '%')\n GROUP BY c.id\n ORDER BY\n CASE WHEN $3 = 'asc' THEN c.id END ASC,\n CASE WHEN $3 = 'desc' THEN c.id END DESC\n OFFSET $1 LIMIT $2 ",
+ "describe": {
+ "columns": [
+ {
+ "ordinal": 0,
+ "name": "id",
+ "type_info": "Int4"
+ },
+ {
+ "ordinal": 1,
+ "name": "cancelled",
+ "type_info": "Bool"
+ },
+ {
+ "ordinal": 2,
+ "name": "complete",
+ "type_info": "Bool"
+ },
+ {
+ "ordinal": 3,
+ "name": "errored",
+ "type_info": "Bool"
+ },
+ {
+ "ordinal": 4,
+ "name": "created_at",
+ "type_info": "Timestamptz"
+ },
+ {
+ "ordinal": 5,
+ "name": "job_ids",
+ "type_info": "Int4Array"
+ },
+ {
+ "ordinal": 6,
+ "name": "message",
+ "type_info": "Varchar"
+ }
+ ],
+ "parameters": {
+ "Left": [
+ "Int8",
+ "Int8",
+ "Text",
+ "Bool",
+ "Text"
+ ]
+ },
+ "nullable": [
+ false,
+ false,
+ false,
+ false,
+ false,
+ null,
+ false
+ ]
+ }
+ },
  "9a4c05da9d9233e6b3fa63ca2f50cf90feb0c305b1cc05e0eb2edcf2572db4ba": {
  "query": "select * from chroma_core_volume where not_deleted = 't'",
  "describe": {

vagrant/Vagrantfile

@@ -133,6 +133,7 @@ Vagrant.configure('2') do |config|
  adm.vm.network 'forwarded_port', guest: 5432, host: 8432
  adm.vm.network 'forwarded_port', guest: 443, host: 8443
  adm.vm.network 'forwarded_port', guest: 7443, host: 7443
+ adm.vm.network 'forwarded_port', guest: 5672, host: 8672
 
  # Admin / management network
  provision_mgmt_net adm, '10'