Skip to content

Keeper: Logging Improvements for Loki #65

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aoikurokawa wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Keeper: Logging Improvements for Loki #65

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3e12ccd
fix: format
aoikurokawa Jul 24, 2026
8c310bb
fix: audit
aoikurokawa Jul 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion .cargo/audit.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,5 +6,7 @@ ignore = [
"RUSTSEC-2025-0022",
"RUSTSEC-2026-0104",
"RUSTSEC-2026-0098",
"RUSTSEC-2026-0099"
"RUSTSEC-2026-0099",
"RUSTSEC-2026-0195",
"RUSTSEC-2026-0194"
]
12 changes: 6 additions & 6 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

15 changes: 6 additions & 9 deletions bam-writer-service/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,16 +106,15 @@ impl BamWriterService {

if let Some(ref overrides) = override_eligible_validators {
log::info!(
"Using override eligible validators ({} pubkeys): {:?}",
"Using override eligible validators count={} validators={:?}",
overrides.len(),
overrides
);
}

if let Some(delegation) = override_delegation_lamports {
log::info!(
"Using override delegation amount: {} lamports ({:.2} SOL)",
delegation,
"Using override delegation amount lamports={delegation} sol={:.2}",
delegation as f64 / 1_000_000_000.0
);
}
Expand Down Expand Up @@ -191,10 +190,9 @@ impl BamWriterService {
Ok(vote_account) => vote_account,
Err(err) => {
log::warn!(
"Skipping historical BAM validator with invalid vote account {} at epoch {}: {}",
"Skipping historical BAM validator with invalid vote account vote_account={} epoch={}: {err:#}",
validator.get_vote_account(),
epoch,
err
);
continue;
}
Expand Down Expand Up @@ -231,7 +229,7 @@ impl BamWriterService {
{
// Override mode: use hardcoded list of vote account pubkeys
log::info!(
"Using override eligible validators ({} pubkeys)",
"Using override eligible validators count={}",
override_validators.len()
);

Expand Down Expand Up @@ -259,7 +257,7 @@ impl BamWriterService {
validators.push(bam_validator);
} else {
log::warn!(
"Override validator {} not found in current vote accounts",
"Override validator not found in current vote accounts vote_pubkey={}",
vote_pubkey
);
}
Expand Down Expand Up @@ -434,8 +432,7 @@ impl BamWriterService {
{
// Override mode: use hardcoded delegation amount
log::info!(
"Using override delegation: {} lamports ({:.2} SOL)",
override_delegation,
"Using override delegation lamports={override_delegation} sol={:.2}",
override_delegation as f64 / 1_000_000_000.0
);

Expand Down
6 changes: 3 additions & 3 deletions bam-writer-service/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,12 +128,12 @@ async fn main() -> anyhow::Result<()> {
if progress >= threshold && !thresholds_hit.contains(&idx) {
let threshold_pct = threshold * 100.0;

info!("Reached {threshold_pct:.0}% threshold for epoch {current_epoch}");
info!("Reached epoch progress threshold_pct={threshold_pct:.0} epoch={current_epoch}");

match bam_writer_service.run().await {
Ok(()) => {
info!(
"Successfully processed at {threshold_pct:.0}% of epoch {current_epoch}"
"BAM writer run succeeded threshold_pct={threshold_pct:.0} epoch={current_epoch}"
);
thresholds_hit.insert(idx);

Expand All @@ -147,7 +147,7 @@ async fn main() -> anyhow::Result<()> {
}
Err(e) => {
error!(
"Error processing at {threshold_pct:.0}% of epoch {current_epoch}: {e}"
"BAM writer run failed threshold_pct={threshold_pct:.0} epoch={current_epoch}: {e:#}"
);

datapoint_info!(
Expand Down
6 changes: 3 additions & 3 deletions cranker/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,14 +48,14 @@ async fn update_stake_pool(config: &Config, epoch: Epoch) -> anyhow::Result<()>
let slack_message = match &result {
Ok(()) => "Cranker has successfully run Stake Pool Update",
Err(e) => {
error!("Cranker failed to update, {e:?}");
error!("Cranker failed to update: {e:#}");
"Cranker failed to update. Please manually run stake pool update"
}
};

// Fire-and-forget Slack notification (don't fail the function if Slack fails)
if let Err(e) = post_slack_message(config.slack_api_token.clone(), slack_message) {
warn!("Slack message failed to post, {e:?}");
warn!("Slack message failed to post: {e:#}");
}

result
Expand All @@ -80,7 +80,7 @@ async fn wait_for_epoch_rewards_completion(config: &Config) -> anyhow::Result<()
tokio::time::sleep(POLL_INTERVAL).await;
}
Err(e) => {
warn!("Failed to check epoch rewards status: {e:?}, retrying...");
warn!("Failed to check epoch rewards status, retrying: {e:#}");
tokio::time::sleep(POLL_INTERVAL).await;
}
}
Expand Down
24 changes: 13 additions & 11 deletions cranker/src/utils.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ pub async fn parallel_execute_transactions(
}
Some(_) | None => {
// If the transaction failed for any other reason, log the error but don't try to resubmit
error!("Transaction failed preflight with err: {e:?}");
error!("Transaction failed preflight: {e:#}");
}
},
}
Expand Down Expand Up @@ -196,7 +196,7 @@ pub async fn parallel_execute_transactions(
}

info!(
"{} transactions submitted, {} confirmed",
"transactions submitted={} confirmed={}",
results.len(),
results.len() - executed_signatures.len()
);
Expand Down Expand Up @@ -263,10 +263,10 @@ pub async fn parallel_execute_stake_pool_update(
);

info!(
"Update list instructions len: {}",
"update_list_instructions={}",
update_list_instructions.len()
);
info!("Final instructions len: {}", final_instructions.len());
info!("final_instructions={}", final_instructions.len());

// Priority fee constants
const INITIAL_PRIORITY_FEE: u64 = 10_000;
Expand All @@ -290,7 +290,7 @@ pub async fn parallel_execute_stake_pool_update(
if let Err(e) =
parallel_execute_transactions(update_list_transactions_prio_fee, config, 250).await
{
error!("Failed to submit update list transactions with initial priority fee: {e:?}");
error!("Failed to submit update list transactions with initial priority fee: {e:#}");
info!("Retrying with higher priority fee");

let update_list_transactions_prio_fee: Vec<Vec<Instruction>> = update_list_instructions
Expand Down Expand Up @@ -322,7 +322,7 @@ pub async fn parallel_execute_stake_pool_update(
.map_err(|e| anyhow!("{e}"))?;

if let Err(e) = retry_send_transaction(config, &transaction, 250).await {
error!("Final transaction failed with initial priority fee: {e:?}");
error!("Final transaction failed with initial priority fee: {e:#}");
info!("Retrying with high priority fee");

let mut final_instructions_prio_fee = vec![
Expand Down Expand Up @@ -372,7 +372,7 @@ pub async fn send_transaction(
return Ok(());
}
Ok(Some(Err(e))) => {
error!("Transaction failed: {signature} {e:?}");
error!("Transaction failed signature={signature}: {e:#}");
return Err(e.into());
}
Ok(None) => {
Expand Down Expand Up @@ -401,10 +401,12 @@ pub async fn simulate_transaction(
.await?
.value;
if result.err.is_some() {
error!("Err: {:?}", result.err);
error!("{:?}", result.logs);
error!(
"Simulation failed err={:?} logs={:?}",
result.err, result.logs
);
} else {
info!("Ok");
info!("Simulation succeeded");
}
Ok(())
}
Expand All @@ -427,7 +429,7 @@ pub async fn retry_send_transaction(
if result.is_ok() {
return result;
} else {
error!("Hit error {result:?}");
error!("Transaction send failed: {result:#?}");
}
}
result
Expand Down
38 changes: 18 additions & 20 deletions steward-writer-service/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ use jito_steward::{
};
use kobe_core::db_models::steward_events::{StewardEvent, StewardEventsStore};
use kobe_core::rpc_utils::{retry_get_slot, retry_get_transactions};
use log::{debug, error, info};
use log::{debug, error, info, warn};
use mongodb::{Client, Collection};
use solana_client::{
nonblocking::rpc_client::RpcClient, rpc_client::GetConfirmedSignaturesForAddress2Config,
Expand Down Expand Up @@ -99,7 +99,7 @@ async fn main() {

match args.command {
Commands::Listen => {
info!("Listening for new events");
info!("Listening for new steward events");
let polling_duration = Duration::from_secs(300); // Configurable polling duration (5 mins)
loop {
if let Err(e) = listen(
Expand All @@ -113,22 +113,22 @@ async fn main() {
)
.await
{
error!("Error in listen loop: {e:?}");
error!("Listen loop failed: {e:#}");
}
}
}
Commands::Backfill {
start_slot,
end_slot,
} => {
info!("Backfilling events from slot {start_slot} to {end_slot:?}");
info!("Backfilling steward events start_slot={start_slot} end_slot={end_slot:?}");
let end_slot = if let Some(end_slot) = end_slot {
end_slot
} else {
match client.get_epoch_info().await {
Ok(epoch_info) => epoch_info.absolute_slot,
Err(e) => {
info!("Error: {e:?}");
error!("Failed to get current slot: {e:#}");
return;
}
}
Expand All @@ -146,7 +146,7 @@ async fn main() {
)
.await
{
info!("Error: {e:?}");
error!("Historical transaction fetch failed: {e:#}");
}
}
}
Expand Down Expand Up @@ -182,7 +182,7 @@ async fn listen(
"cluster" => cluster_name,
);

info!("Fetching new transactions since signature: {latest_signature}");
info!("Fetching new transactions since_signature={latest_signature}");

let mut before = None;

Expand All @@ -203,7 +203,7 @@ async fn listen(
{
Ok(signatures) => signatures,
Err(e) => {
info!("Error fetching RPC signatures: {e}");
warn!("Failed to fetch RPC signatures, retrying: {e:#}");
continue;
}
};
Expand Down Expand Up @@ -244,7 +244,7 @@ async fn fetch_and_process_transactions(

let transactions = retry_get_transactions(rpc_client, &transaction_signatures).await?;

info!("Fetched {} transactions from rpc", transactions.len());
info!("Fetched transactions count={}", transactions.len());

let mut transaction_data = vec![];
for tx in transactions.into_iter() {
Expand Down Expand Up @@ -278,7 +278,7 @@ async fn process_transactions(
for (status, tx) in transactions.iter() {
if tx.slot != status.slot {
error!(
"Slot mismatch for signature {}: {} != {}",
"Slot mismatch signature={} tx_slot={} status_slot={}",
status.signature, tx.slot, status.slot
);
}
Expand Down Expand Up @@ -335,11 +335,9 @@ async fn process_transactions(
{
Ok(Some(event)) => events.push(event),
Ok(None) => {}
Err(e) => error!(
"Error parsing log message for transaction {:?}: {:?}",
signature,
e.to_string()
),
Err(e) => {
error!("Failed to parse log message signature={signature:?}: {e:#}")
}
}
}
}
Expand All @@ -348,11 +346,11 @@ async fn process_transactions(

match dry_run {
true => {
info!("upserting {events:#?}");
info!("Dry run upserting events count={}", events.len());
}
false => {
if let Err(e) = store.bulk_upsert(events).await {
error!("Error inserting events: {e:?}");
error!("Failed to insert events: {e:#}");
}
}
}
Expand All @@ -371,7 +369,7 @@ async fn fetch_historical_program_transactions(
end_slot: u64,
dry_run: bool,
) -> Result<(), Box<dyn std::error::Error>> {
info!("Backfilling transactions between slots {start_slot} and {end_slot}");
info!("Backfilling transactions start_slot={start_slot} end_slot={end_slot}");
let mut before = None;
let mut should_break = false;

Expand All @@ -390,7 +388,7 @@ async fn fetch_historical_program_transactions(
{
Ok(signatures) => signatures,
Err(e) => {
info!("Error fetching RPC signatures: {e}");
warn!("Failed to fetch RPC signatures, retrying: {e:#}");
continue;
}
};
Expand Down Expand Up @@ -424,7 +422,7 @@ async fn fetch_historical_program_transactions(
}

info!(
"Processing {} transactions starting at slot {}",
"Processing transactions count={} start_slot={}",
valid_signatures.len(),
valid_signatures[0].slot
);
Expand Down
Loading
Loading