To create a kubectl plugin whereby, for example, we could rung a new tool like this

kubectl log index echo 3 -n dev

Where the above would find pods with a partial name of echo and from those pods that match, finds the index 3 (0 indexed).

To create a plugin you use the naming convention

kubeclt-<your-plugin-name>

You need to build your plugin then ensure it’s copied into your PATH.

Once built and copied, you can use the following to check if kubectl can find the plugin

kubectl plugin list

Sample Plugin

I’ve created the plugin using Rust.

Note: This is just a quick implementation and not fully tested, but gives an idea of how to create such a plugin.

Set your Cargo.toml dependencies as follows

k8s-openapi = { version = "0.26.0", features = ["v1_32"] }
kube = { version = "2.0.1", features = ["runtime", "derive"] }
tokio = { version = "1", features = ["full"] }
clap = { version = "4", features = ["derive"] }
anyhow = "1.0"

Next we want to create the command line arguments using the following

#[derive(Parser, Debug)]
#[command(name = "kubectl-log-index")]
#[command(author, version, about)]
pub struct Args {
    /// Partial name of the pod to match
    #[arg()]
    pub pod_part: String,
    /// Index of the pod (0-based)
    pub index: usize,
    /// Follow the log stream
    #[arg(short = 'f', long)]
    pub follow: bool,
    /// Kubernetes namespace (optional)
    #[arg(short, long)]
    pub namespace: Option<String>,
}

We’re supplying some short form parameters such as -f which can be used instead of –follow, likewise -n in place of –namespace.

Our main.rs looks like this

mod args;

use clap::Parser;
use anyhow::Result;
use kube::{Api, Client};
use k8s_openapi::api::core::v1::Pod;
use std::process::Command;
use kube::api::ListParams;
use kube::runtime::reflector::Lookup;
use crate::args::Args;

/// kubectl plugin to get logs by container index
#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();

    let namespace: &str = args.namespace
        .as_deref()
        .unwrap_or("default");

    let client = Client::try_default().await?;
    let pods: Api<Pod> = Api::namespaced(client, namespace);

    let pod_list = find_matching_pods(pods, &args.pod_part).await.expect("Failed to find matching pods");
    
    let pod = pod_list
        .get(args.index)
        .cloned()
        .ok_or_else(|| anyhow::anyhow!("Pod not found"))?;

    let pod_name = &pod.name().ok_or_else(|| anyhow::anyhow!("Pod name not found"))?;

    let mut cmd = Command::new("kubectl");

    cmd.args(["logs", pod_name]);

    if namespace != "default" {
        cmd.args(["-n", namespace]);
    }

    if args.follow {
        cmd.arg("-f");
    }

    cmd
        .status()?;

    Ok(())
}

pub async fn find_matching_pods(
    pods: Api<Pod>,
    partial: &str,
) -> Result<Vec<Pod>, Box<dyn std::error::Error>> {
    let pod_list = pods.list(&ListParams::default()).await?;

    let matches: Vec<Pod> = pod_list.items
        .into_iter()
        .filter(|pod| {
            pod.metadata.name
                .as_ref()
                .map(|name| name.contains(partial))
                .unwrap_or(false)
        })
        .collect();

    Ok(matches)
}