feat(tunnel): show tunnel information

This introduces a new `Tunnel` widget that shows
the status of tunnels in the dashboard. This
allows users to see what's currently happening.
There are some caveats to this though:
- Egress starts up immediately and is `listening`.
- Ingress is only on incoming connections and
  switches to `inactive` when the connection is
  closed.

Additionally, this:
- Standardizes the `Session` "writers" into a
  broadcast that can be either for a single
  channel or all channels.
- No longer hangs indefinitely when a source
  channel is closed (via ingress tunnel) but the
  destination is still open (warp::serve keeps the
  connection open for example).
- Stops trying to close all channels that EOF.
  This is because the tunnel channels now manage
  their lifecycle themselves. Unfortunately, the
  pty and sftp requests do not.
  `Session::channels` now has a `None` inserted if
  a channel has been consumed but it is still
  open.
- Passing a `StringError` that is clone-able
  instead of `Report` which is not. This results
  in the source needing to render the error but it
  makes everything much easier to work with.
- Rewords `StatusError` to no longer exist -
  instead being constructed as a `Report` by
  `PodExt`.
- Moves as much of the verbose logging as possible
  to `debug` from `info`. See `DEVELOPMENT.md` for
  how to enable this.
- Allow `Table` to have an optional header and
  selection style.

Author: grampelberg

.cargo/audit.toml

@@ -1,2 +1,2 @@
 [advisories]
-ignore = ["RUSTSEC-2023-0071", "RUSTSEC-2024-0320"]
+ignore = ["RUSTSEC-2023-0071", "RUSTSEC-2024-0320", "RUSTSEC-2024-0370"]

.github/workflows/signed.yml

@@ -0,0 +1,16 @@
+name: check-signed-commits
+on:
+  pull_request_target:
+    branches:
+      - main
+
+jobs:
+  check-signed-commits:
+    name: verify-signed-commits
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      pull-requests: write
+    steps:
+      - name: Check signed commits in PR
+        uses: 1Password/check-signed-commits-action@v1

DEVELOPMENT.md

@@ -42,6 +42,14 @@ be available to run inside the cluster.
 
 [k3d]: https://k3d.io/v5.6.3/#releases
 
+## Logging
+
+The global debug level can be overly noisy. Instead of doing `-vvvv`, try:
+
+```bash
+RUST_LOG=none,kuberift=debug
+```
+
 ## Ingress Tunnel
 
 If testing port forwarding and running the service locally (aka not on the

README.md

@@ -28,7 +28,6 @@ You can:
   kuberift itself.
 - [Metrics](docs/metrics.md) - List of the possible metrics exported via.
   prometheus.
-- [TODO](TODO.md) - A selection of outstanding functionality.
 
 [auth]: docs/auth.md
 

TODO.md

@@ -54,16 +54,8 @@
 
 ## Ingress Tunnel
 
-- Add an information banner for tunnels that have been requested and are active.
-
 ## Egress Tunnel
 
-- Add an information banner for tunnels that have been requested and are active.
-- Keep the task running even if the first connect fails - need to be able to
-  retry it without restarting the session.
-- Display error for when the `tcpip_forward` address is `localhost`.
-- Need some way to do cleanup and lifecycle management of endpoints and
-  services.
 - Cleanup services/endpoints on:
   - Shutdown - especially termination of the channel.
   - Startup - because we can't do cross-namespace owner references, anything

docs/deployment.md

@@ -51,6 +51,10 @@ using helm, there are some things to be aware of:
   using the [gateway api][gateway-api] or configuring your ingress controller to
   route TCP.
 
+Note: if you're debugging something, instead of setting global verbosity with
+`-vv`, use `RUST_LOG=none,kuberift=debug`. That'll keep other crates that are
+especially noisy out of the output.
+
 ### Helm
 
 There is a provided `getting-started.yaml` set of values. To install this on

src/broadcast.rs

@@ -0,0 +1,43 @@
+use std::{collections::HashMap, sync::Arc};
+
+use eyre::{eyre, Result};
+use russh::ChannelId;
+use tokio::sync::{mpsc::UnboundedSender, Mutex};
+
+use crate::events::Event;
+
+#[derive(Debug, Clone, Default)]
+pub struct Broadcast {
+    channels: Arc<Mutex<HashMap<ChannelId, UnboundedSender<Event>>>>,
+}
+
+impl Broadcast {
+    pub async fn add(&mut self, id: ChannelId, tx: UnboundedSender<Event>) -> Result<()> {
+        self.channels.lock().await.insert(id, tx);
+
+        Ok(())
+    }
+
+    pub async fn remove(&mut self, id: &ChannelId) -> Option<UnboundedSender<Event>> {
+        self.channels.lock().await.remove(id)
+    }
+
+    pub async fn send(&self, id: &ChannelId, event: Event) -> Result<()> {
+        let mut channels = self.channels.lock().await;
+        if let Some(sender) = channels.get_mut(id) {
+            sender
+                .send(event)
+                .map_err(|_| eyre!("failed to send event"))?;
+        }
+        Ok(())
+    }
+
+    pub async fn all(&self, event: Event) -> Result<()> {
+        let mut channels = self.channels.lock().await;
+        for sender in channels.values_mut() {
+            sender.send(event.clone())?;
+        }
+
+        Ok(())
+    }
+}

src/cli/dev/dashboard.rs

@@ -69,8 +69,6 @@ impl AsyncRead for Stdin {
 
             let n = std::io::stdin().read(buf.initialize_unfilled())?;
             buf.advance(n);
-
-            tracing::info!("buf: {:?}", buf.filled());
         }
 
         if !buf.filled().is_empty() {

src/cli/serve.rs

@@ -112,7 +112,7 @@ impl Serve {
         };
 
         let ctrl = ControllerBuilder::default()
-            .current(
+            .server(
                 CurrentPodBuilder::default()
                     .namespace(cfg.default_namespace.clone())
                     .name(self.pod_name.clone())

src/dashboard.rs

@@ -12,7 +12,7 @@ use tokio::{
 use tokio_util::io::ReaderStream;
 
 use crate::{
-    events::{Broadcast, Event, Input, Keypress},
+    events::{Broadcast, Event, Input, Keypress, StringError},
     io::{backend::Backend, Writer},
     widget::{apex::Apex, Raw, Widget},
 };
@@ -185,7 +185,9 @@ async fn run(
                 let raw_result =
                     draw_raw(raw_widget, &mut term, &mut rx, stdout.non_blocking()).await;
 
-                let result = current_widget.dispatch(&Event::Finished(raw_result))?;
+                let result = current_widget.dispatch(&Event::Finished(
+                    raw_result.map_err(|e| StringError(format!("{e:?}"))),
+                ))?;
 
                 state.ui();
 

src/events.rs

@@ -1,10 +1,10 @@
-use std::str;
+use std::{error::Error, str};
 
 use eyre::Result;
 use ratatui::backend::WindowSize;
 use tokio_util::bytes::Bytes;
 
-use crate::widget::Raw;
+use crate::{resources::tunnel, widget::Raw};
 
 #[derive(Debug)]
 pub enum Broadcast {
@@ -14,15 +14,15 @@ pub enum Broadcast {
     Raw(Box<dyn Raw>),
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum Event {
     Input(Input),
     Resize(WindowSize),
     Goto(Vec<String>),
-    Error(String),
     Shutdown,
     Render,
-    Finished(Result<()>),
+    Finished(Result<(), StringError>),
+    Tunnel(Result<tunnel::Tunnel, tunnel::Error>),
 }
 
 impl Event {
@@ -49,7 +49,18 @@ impl From<Bytes> for Event {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
+pub struct StringError(pub String);
+
+impl Error for StringError {}
+
+impl std::fmt::Display for StringError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+#[derive(Debug, Clone)]
 pub struct Input {
     pub key: Keypress,
     raw: Bytes,

src/main.rs

@@ -1,5 +1,6 @@
 //! # kuberift
 
+mod broadcast;
 #[warn(dead_code)]
 mod cli;
 mod dashboard;

src/resources.rs

@@ -2,6 +2,7 @@ pub mod age;
 pub mod container;
 pub mod file;
 pub mod pod;
+pub mod status;
 pub mod store;
 pub mod tunnel;
 
@@ -18,7 +19,7 @@ use kube::{
 };
 use regex::Regex;
 use serde::Serialize;
-use tracing::info;
+pub use tunnel::Tunnel;
 
 use crate::identity;
 
@@ -32,11 +33,11 @@ pub(crate) async fn create(
     client: &Api<CustomResourceDefinition>,
     update: bool,
 ) -> Result<Vec<CustomResourceDefinition>> {
-    info!(update = update, "updating CRD definitions...");
+    tracing::info!(update = update, "updating CRD definitions...");
 
     let results: Vec<_> = futures::stream::iter(all())
         .map(|resource| async move {
-            info!("creating/updating CRD: {}", resource.name_any());
+            tracing::info!("creating/updating CRD: {}", resource.name_any());
 
             if update {
                 client

src/resources/container.rs

@@ -197,15 +197,15 @@ impl ContainerExt for Container {
 }
 
 impl<'a> TableRow<'a> for Container {
-    fn header() -> Row<'a> {
-        Row::new(vec![
+    fn header() -> Option<Row<'a>> {
+        Some(Row::new(vec![
             Cell::from("Name"),
             Cell::from("Image"),
             Cell::from("Ready"),
             Cell::from("State"),
             Cell::from("Restarts"),
             Cell::from("Age"),
-        ])
+        ]))
     }
 
     fn constraints() -> Vec<Constraint> {

src/resources/pod.rs

@@ -1,14 +1,11 @@
 pub mod proc;
 
-use std::{borrow::Borrow, cmp::Ordering, error::Error, fmt::Display, net::IpAddr, sync::Arc};
+use std::{borrow::Borrow, cmp::Ordering, net::IpAddr, sync::Arc};
 
 use chrono::{TimeDelta, Utc};
-use k8s_openapi::{
-    api::core::v1::{
-        ContainerState, ContainerStateTerminated, ContainerStateWaiting, ContainerStatus, Pod,
-        PodStatus,
-    },
-    apimachinery::pkg::apis::meta::v1,
+use k8s_openapi::api::core::v1::{
+    ContainerState, ContainerStateTerminated, ContainerStateWaiting, ContainerStatus, Pod,
+    PodStatus,
 };
 use kube::ResourceExt;
 pub use proc::Proc;
@@ -27,43 +24,6 @@ use crate::widget::{
     TableRow,
 };
 
-// TODO: There's probably a better debug implementation than this.
-#[derive(Clone, Debug)]
-pub struct StatusError {
-    pub message: String,
-}
-
-// Because this is a golang error that's being returned, there's really no good
-// way to convert this into something that is moderately usable. The rest of the
-// `Status` struct is empty of anything useful. The decision is to be naive here
-// and let other display handlers figure out if they would like to deal with the
-// message.
-impl StatusError {
-    pub fn new(inner: v1::Status) -> Self {
-        Self {
-            message: inner.message.unwrap_or("unknown status".to_string()),
-        }
-    }
-}
-
-impl Error for StatusError {}
-
-impl Display for StatusError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.message)
-    }
-}
-
-pub trait StatusExt {
-    fn is_success(&self) -> bool;
-}
-
-impl StatusExt for v1::Status {
-    fn is_success(&self) -> bool {
-        self.status == Some("Success".to_string())
-    }
-}
-
 pub enum Phase {
     Pending,
     Running,
@@ -268,15 +228,15 @@ impl PodExt for Pod {
 }
 
 impl<'a> TableRow<'a> for Arc<Pod> {
-    fn header() -> Row<'a> {
-        Row::new(vec![
+    fn header() -> Option<Row<'a>> {
+        Some(Row::new(vec![
             Cell::from("Namespace"),
             Cell::from("Name"),
             Cell::from("Ready"),
             Cell::from("Status"),
             Cell::from("Restarts"),
             Cell::from("Age"),
-        ])
+        ]))
     }
 
     fn constraints() -> Vec<Constraint> {

src/resources/pod/proc.rs

@@ -3,8 +3,10 @@ use k8s_openapi::api::core::v1::Pod;
 use kube::api::{Api, AttachParams};
 use tokio::io::AsyncReadExt;
 
-use super::{StatusError, StatusExt};
-use crate::resources::container::{Container, ContainerExt};
+use crate::resources::{
+    container::{Container, ContainerExt},
+    status::StatusExt,
+};
 
 pub struct Proc {
     container: Container,
@@ -47,7 +49,7 @@ impl Proc {
 
         if let Some(status) = status.await {
             if !status.is_success() {
-                return Err(eyre!(StatusError::new(status)));
+                return Err(status.into_report());
             }
         }