Add restatectl logs trim subcommand

crates/admin/src/cluster_controller/service.rs

@@ -18,7 +18,7 @@ use futures::{Stream, StreamExt};
 use tokio::sync::{mpsc, oneshot};
 use tokio::time;
 use tokio::time::{Instant, Interval, MissedTickBehavior};
-use tracing::{debug, warn};
+use tracing::{debug, info, warn};
 
 use restate_bifrost::{Bifrost, BifrostAdmin};
 use restate_core::metadata_store::MetadataStoreClient;
@@ -335,7 +335,10 @@ where
                 trim_point,
                 response_tx,
             } => {
-                debug!("Manual trim log '{log_id}' until (inclusive) lsn='{trim_point}'");
+                info!(
+                    ?log_id,
+                    trim_point_inclusive = ?trim_point,
+                    "Manual trim log command received");
                 let result = bifrost_admin.trim(log_id, trim_point).await;
                 let _ = response_tx.send(result.map_err(Into::into));
             }

crates/bifrost/src/providers/local_loglet/mod.rs

@@ -248,6 +248,13 @@ impl Loglet for LocalLoglet {
 
         histogram!(BIFROST_LOCAL_TRIM_LENGTH).record(*effective_trim_point - *current_trim_point);
 
+        debug!(
+            loglet_id = self.loglet_id,
+            ?current_trim_point,
+            ?effective_trim_point,
+            "Loglet trim operation enqueued"
+        );
+
         Ok(())
     }
 

tools/restatectl/src/commands/log/mod.rs

@@ -11,6 +11,7 @@
 mod describe_log;
 mod dump_log;
 mod list_logs;
+mod trim_log;
 
 use cling::prelude::*;
 
@@ -22,4 +23,6 @@ pub enum Log {
     Describe(describe_log::DescribeLogIdOpts),
     /// Dump the contents of a bifrost log
     Dump(dump_log::DumpLogOpts),
+    /// Trim a log to a particular Log Sequence Number (LSN)
+    Trim(trim_log::TrimLogOpts),
 }

tools/restatectl/src/commands/log/trim_log.rs

@@ -0,0 +1,60 @@
+// Copyright (c) 2024 -  Restate Software, Inc., Restate GmbH.
+// All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+use anyhow::Context;
+use cling::prelude::*;
+use tonic::codec::CompressionEncoding;
+
+use restate_admin::cluster_controller::protobuf::cluster_ctrl_svc_client::ClusterCtrlSvcClient;
+use restate_admin::cluster_controller::protobuf::TrimLogRequest;
+use restate_cli_util::c_println;
+
+use crate::app::ConnectionInfo;
+use crate::util::grpc_connect;
+
+#[derive(Run, Parser, Collect, Clone, Debug)]
+#[clap()]
+#[cling(run = "trim_log")]
+pub struct TrimLogOpts {
+    /// The log to trim
+    #[arg(short, long)]
+    log_id: u32,
+
+    /// The Log Sequence Number (LSN) to trim the log to, inclusive
+    #[arg(short, long)]
+    trim_point: u64,
+}
+
+async fn trim_log(connection: &ConnectionInfo, opts: &TrimLogOpts) -> anyhow::Result<()> {
+    let channel = grpc_connect(connection.cluster_controller.clone())
+        .await
+        .with_context(|| {
+            format!(
+                "cannot connect to cluster controller at {}",
+                connection.cluster_controller
+            )
+        })?;
+    let mut client =
+        ClusterCtrlSvcClient::new(channel).accept_compressed(CompressionEncoding::Gzip);
+
+    let trim_request = TrimLogRequest {
+        log_id: opts.log_id,
+        trim_point: opts.trim_point,
+    };
+    client
+        .trim_log(trim_request)
+        .await
+        .with_context(|| "failed to submit trim request")?
+        .into_inner();
+
+    c_println!("Submitted");
+
+    Ok(())
+}