Correlate Kubernetes pod traffic logs with other pods in the cluster by IP address

app.js

import Kubernetes from './kubernetes.js'

const kubectl = new Kubernetes({namespace: process.env.KUBE_NAMESPACE});
const logMinutes = 30;
(async () => {

  // Tally IP addresses that hit us
  const ipHits = new Map;
  for (const pod of await kubectl.listPods({app: process.env.KUBE_APP_LABEL})) {
    console.log(`Checking ${logMinutes}m of`, pod.metadata.name)
    for (const line of await kubectl.getRecentLogLines(pod.metadata.name, {
      maxSeconds: logMinutes*60
    })) {
      // parse typical HTTP access log
      if (!line.includes(`- -`)) continue;
      const ip = line.split(' ')[0];
      ipHits.set(ip, (ipHits.get(ip) || 0) + 1);
    }
  }

  // Correlate with pods from the cluster
  const seenPods = new Map;
  for (const pod of await kubectl.listGlobalPods()) {
    if (!ipHits.has(pod.status.podIP)) continue;
    seenPods.set(pod.status.podIP, {
      namespace: pod.metadata.namespace,
      name: pod.metadata.name,
      hits: ipHits.get(pod.status.podIP),
    });
    ipHits.delete(pod.status.podIP);
  }

  console.log('Pod traffic:', seenPods);
  console.log('Unaccounted traffic:', ipHits);
})();

kubernetes.js

import k8s from '@kubernetes/client-node'

export default class KubernetesClient {
  constructor({context, namespace}) {
    this.namespace = namespace;

    this.config = new k8s.KubeConfig();
    this.config.loadFromDefault();
    if (context) this.config.setCurrentContext(context);

    this.coreApi = this.config.makeApiClient(k8s.CoreV1Api);
  }

  async listPods(labels={}) {
    const resp = await this.coreApi.listNamespacedPod(this.namespace, undefined, undefined, undefined, labels)
    return resp.body.items;
  }

  async getRecentLogLines(podName, {container, maxSeconds, maxBytes}) {
    const resp = await this.coreApi.readNamespacedPodLog(podName, this.namespace,
      container, false, false, maxBytes, false, false, maxSeconds);
    return (resp.body||'').split('\n').slice(0, -1);
  }

  async listGlobalPods() {
    const resp = await this.coreApi.listPodForAllNamespaces();
    return resp.body.items;
  }
};