Add db subcommand

daqingest/src/bin/daqingest.rs

@@ -3,6 +3,8 @@ use daqingest::opts::DaqIngestOpts;
 use err::Error;
 use log::*;
 use netfetch::conf::parse_config;
+use netpod::Database;
+use scywr::config::ScyllaIngestConfig;
 use taskrun::TracingMode;
 
 pub fn main() -> Result<(), Error> {
@@ -46,6 +48,22 @@ async fn main_run_inner(opts: DaqIngestOpts) -> Result<(), Error> {
             let scylla_conf = err::todoval();
             scywr::tools::fetch_events(&k.backend, &k.channel, &scylla_conf).await?
         }
+        SubCmd::Db(k) => {
+            use daqingest::opts::DbSub;
+            let pgconf = Database {
+                host: k.pg_host,
+                port: k.pg_port,
+                user: k.pg_user,
+                pass: k.pg_pass,
+                name: k.pg_name,
+            };
+            let scyconf = ScyllaIngestConfig::new([k.scylla_host], k.scylla_keyspace);
+            match k.sub {
+                DbSub::RemoveOlder(j) => {
+                    info!("RemoveOlder  {:?}  {:?}", pgconf, scyconf);
+                }
+            }
+        }
         SubCmd::ChannelAccess(k) => match k {
             #[cfg(DISABLED)]
             ChannelAccess::CaSearch(k) => {

daqingest/src/lib.rs

@@ -1,3 +1,3 @@
 pub mod daemon;
 pub mod opts;
-pub mod query;
+pub mod tools;

daqingest/src/opts.rs

@@ -22,6 +22,7 @@ pub enum SubCmd {
     ListPkey,
     ListPulses,
     FetchEvents(FetchEvents),
+    Db(Db),
     #[command(subcommand)]
     ChannelAccess(ChannelAccess),
     #[cfg(feature = "bsread")]
@@ -92,3 +93,35 @@ pub struct CaSearch {
 pub struct CaConfig {
     pub config: String,
 }
+
+#[derive(Debug, clap::Parser)]
+pub struct Db {
+    #[arg(long)]
+    pub scylla_host: String,
+    #[arg(long)]
+    pub scylla_keyspace: String,
+    #[arg(long)]
+    pub pg_host: String,
+    #[arg(long)]
+    #[clap(default_value = "5432")]
+    pub pg_port: u16,
+    #[arg(long)]
+    pub pg_user: String,
+    #[arg(long)]
+    pub pg_pass: String,
+    #[arg(long)]
+    pub pg_name: String,
+    #[command(subcommand)]
+    pub sub: DbSub,
+}
+
+#[derive(Debug, clap::Parser)]
+pub enum DbSub {
+    RemoveOlder(RemoveOlder),
+}
+
+#[derive(Debug, clap::Parser)]
+pub struct RemoveOlder {
+    #[arg(long)]
+    date: String,
+}

daqingest/src/query.rs

@@ -1 +0,0 @@
-

netfetch/src/ca/findioc.rs

@@ -689,7 +689,7 @@ impl Stream for FindIocStream {
             }
             break match self.afd.poll_read_ready(cx) {
                 Ready(Ok(mut g)) => {
-                    debug!("BLOCK AA");
+                    // debug!("BLOCK AA");
                     match unsafe { Self::try_read(self.sock.0, &self.stats) } {
                         Ready(Ok((src, res))) => {
                             self.handle_result(src, res);