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 1 commit
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
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: 10 additions & 14 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 @@
}
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 @@
}

info!(
"{} transactions submitted, {} confirmed",
"transactions submitted={} confirmed={}",
results.len(),
results.len() - executed_signatures.len()
);
Expand Down Expand Up @@ -259,14 +259,11 @@
&stake_pool,
&validator_list,
&config.stake_pool_address,
no_merge,

Check warning on line 262 in cranker/src/utils.rs

View workflow job for this annotation

GitHub Actions / lint / lint 

Diff in /home/runner/work/kobe/kobe/cranker/src/utils.rs
);

info!(
"Update list instructions len: {}",
update_list_instructions.len()
);
info!("Final instructions len: {}", final_instructions.len());
info!("update_list_instructions={}", update_list_instructions.len());
info!("final_instructions={}", final_instructions.len());
Comment thread
aoikurokawa marked this conversation as resolved.
Outdated

// Priority fee constants
const INITIAL_PRIORITY_FEE: u64 = 10_000;
Expand All @@ -290,7 +287,7 @@
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 +319,7 @@
.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 +369,7 @@
return Ok(());
}
Ok(Some(Err(e))) => {
error!("Transaction failed: {signature} {e:?}");
error!("Transaction failed signature={signature}: {e:#}");
return Err(e.into());
}
Ok(None) => {
Expand All @@ -398,13 +395,12 @@
let result = config
.rpc_client
.simulate_transaction(&transaction)
.await?

Check warning on line 398 in cranker/src/utils.rs

View workflow job for this annotation

GitHub Actions / lint / lint 

Diff in /home/runner/work/kobe/kobe/cranker/src/utils.rs
.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 +423,7 @@
if result.is_ok() {
return result;
} else {
error!("Hit error {result:?}");
error!("Transaction send failed: {result:#?}");
}
}
result
Expand Down
34 changes: 16 additions & 18 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 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 @@

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 @@
)
.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 @@
)
.await
{
info!("Error: {e:?}");
error!("Historical transaction fetch failed: {e:#}");
}
}
}
Expand Down Expand Up @@ -182,7 +182,7 @@
"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 @@
{
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 @@

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 @@
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 @@ -332,13 +332,11 @@
*slot,
)
.await
{

Check warning on line 335 in steward-writer-service/src/main.rs

View workflow job for this annotation

GitHub Actions / lint / lint 

Diff in /home/runner/work/kobe/kobe/steward-writer-service/src/main.rs
Ok(Some(event)) => events.push(event),
Ok(None) => {}
Err(e) => error!(
"Error parsing log message for transaction {:?}: {:?}",
signature,
e.to_string()
"Failed to parse log message signature={signature:?}: {e:#}"
),
}
}
Expand All @@ -348,11 +346,11 @@

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 @@
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 @@
{
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 @@
}

info!(
"Processing {} transactions starting at slot {}",
"Processing transactions count={} start_slot={}",
valid_signatures.len(),
valid_signatures[0].slot
);
Expand Down
13 changes: 8 additions & 5 deletions writer-service/src/bam_boost_manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,28 +88,31 @@ impl BamBoostManager {
"https://storage.googleapis.com/jito-bam-boost/{network}/{epoch}/merkle_tree.json",
);

log::info!("Fetching merkle tree from: {url}");
log::info!("Fetching bam boost merkle tree url={url}");

// Download the merkle tree JSON from GCS
let response = match reqwest::get(&url).await {
Ok(resp) => resp,
Err(e) => {
log::error!("Failed to fetch merkle tree: {e}");
log::error!("Failed to fetch bam boost merkle tree url={url}: {e:#}");
return Err(AppError::FileNotFound(
"Failed to fetch merkle tree ({url}: {e}".to_string(),
));
Comment thread
aoikurokawa marked this conversation as resolved.
}
};

if !response.status().is_success() {
log::error!("Merkle tree not found: status {}", response.status());
log::error!(
"Bam boost merkle tree not found url={url} status={}",
response.status()
);
return Err(AppError::InvalidOperation(format!(
"Merkle tree not found for network {network} epoch {epoch}",
)));
}

let response_json: Vec<BamBoostEntry> = response.json().await.map_err(|e| {
log::error!("Failed to parse merkle tree JSON: {e}");
log::error!("Failed to parse bam boost merkle tree JSON url={url}: {e:#}");
AppError::InvalidOperation(format!(
"Failed to parse merkle tree JSON for network {network} epoch {epoch}: {e}",
))
Expand Down Expand Up @@ -175,7 +178,7 @@ impl BamBoostManager {
}
}
Err(e) => {
log::info!("Skip fetching bam boost merkle tree {epoch}: {e}");
log::warn!("Skipping bam boost merkle tree fetch epoch={epoch}: {e:#}");
}
}
}
Expand Down
Loading
Loading