k3s-gitops/apps/ollama-mcp/mcp-loki/index.js

const express = require('express');
const axios = require('axios');

const app = express();
const port = process.env.PORT || 3000;

const LOKI_URL = (process.env.LOKI_URL || 'http://localhost:3100').replace(/\/$/, '');

app.use(express.json());
app.use((req, res, next) => {
  res.header('Access-Control-Allow-Origin', '*');
  res.header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS');
  res.header('Access-Control-Allow-Headers', 'Content-Type, Authorization');
  if (req.method === 'OPTIONS') return res.sendStatus(200);
  next();
});

// ─── Helpers ────────────────────────────────────────────────────────────────

const nowNs = () => (Date.now() * 1_000_000).toString();
const hoursAgoNs = (h) => ((Date.now() - h * 3600 * 1000) * 1_000_000).toString();
const minutesAgoNs = (m) => ((Date.now() - m * 60 * 1000) * 1_000_000).toString();

const lokiQuery = async (query, start, end, limit, direction) => {
  const params = {
    query,
    start: start || hoursAgoNs(1),
    end: end || nowNs(),
    limit: limit || 100,
    direction: direction || 'backward',
  };
  const r = await axios.get(`${LOKI_URL}/loki/api/v1/query_range`, { params });
  return r.data;
};

const lokiInstant = async (query, time) => {
  const params = { query, time: time || nowNs() };
  const r = await axios.get(`${LOKI_URL}/loki/api/v1/query`, { params });
  return r.data;
};

const flattenStreams = (data) => {
  const lines = [];
  for (const stream of data?.data?.result || []) {
    const labels = stream.stream;
    for (const [ts, line] of stream.values || []) {
      lines.push({ timestamp: new Date(parseInt(ts) / 1_000_000).toISOString(), labels, line });
    }
  }
  return lines.sort((a, b) => new Date(b.timestamp) - new Date(a.timestamp));
};

// ─── Routes ─────────────────────────────────────────────────────────────────

app.get('/health', (req, res) => res.json({ status: 'healthy', service: 'mcp-loki', loki: LOKI_URL }));

app.get('/', (req, res) => res.json({
  service: 'MCP Loki Server',
  version: '1.0.0',
  loki: LOKI_URL,
  endpoints: [
    'POST /api/query          - Raw LogQL query with time range',
    'POST /api/labels         - List all available label names',
    'POST /api/label_values   - List values for a specific label',
    'POST /api/namespaces     - List all Kubernetes namespaces with logs',
    'POST /api/pods           - List pods with logs (optionally filter by namespace)',
    'POST /api/pod_logs       - Get logs for a specific pod',
    'POST /api/namespace_logs - Get recent logs for all pods in a namespace',
    'POST /api/errors         - Find ERROR/WARN log lines across cluster or namespace',
    'POST /api/search         - Full-text search across logs',
    'POST /api/oomkilled      - Find OOMKilled events cluster-wide',
    'POST /api/crash_loops    - Find CrashLoopBackOff events',
    'POST /api/rate           - Log ingestion rate by namespace',
    'POST /api/stats          - Loki server stats and ingestion metrics',
  ]
}));

// Raw LogQL query
app.post('/api/query', async (req, res) => {
  try {
    const { query, start, end, limit, direction } = req.body;
    if (!query) return res.status(400).json({ error: 'query (LogQL) is required' });
    const data = await lokiQuery(query, start, end, limit, direction);
    const lines = flattenStreams(data);
    res.json({ count: lines.length, lines });
  } catch (e) {
    res.status(500).json({ error: e.message, loki: LOKI_URL });
  }
});

// List label names
app.post('/api/labels', async (req, res) => {
  try {
    const r = await axios.get(`${LOKI_URL}/loki/api/v1/labels`);
    res.json({ labels: r.data.data });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// List values for a label
app.post('/api/label_values', async (req, res) => {
  try {
    const { label } = req.body;
    if (!label) return res.status(400).json({ error: 'label is required' });
    const r = await axios.get(`${LOKI_URL}/loki/api/v1/label/${label}/values`);
    res.json({ label, values: r.data.data });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// List namespaces with logs
app.post('/api/namespaces', async (req, res) => {
  try {
    const r = await axios.get(`${LOKI_URL}/loki/api/v1/label/namespace/values`);
    res.json({ namespaces: r.data.data });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// List pods (optionally by namespace)
app.post('/api/pods', async (req, res) => {
  try {
    const { namespace } = req.body;
    let url = `${LOKI_URL}/loki/api/v1/label/pod/values`;
    if (namespace) url += `?query={namespace="${namespace}"}`;
    const r = await axios.get(url);
    res.json({ namespace: namespace || 'all', pods: r.data.data });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// Pod logs
app.post('/api/pod_logs', async (req, res) => {
  try {
    const { pod, namespace, container, minutes, limit } = req.body;
    if (!pod) return res.status(400).json({ error: 'pod is required' });

    let query = `{pod="${pod}"`;
    if (namespace) query += `,namespace="${namespace}"`;
    if (container) query += `,container="${container}"`;
    query += '}';

    const start = minutesAgoNs(minutes || 30);
    const data = await lokiQuery(query, start, nowNs(), limit || 200);
    const lines = flattenStreams(data);
    res.json({ pod, namespace, container, count: lines.length, lines });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// All logs in a namespace
app.post('/api/namespace_logs', async (req, res) => {
  try {
    const { namespace, minutes, limit } = req.body;
    if (!namespace) return res.status(400).json({ error: 'namespace is required' });

    const query = `{namespace="${namespace}"}`;
    const start = minutesAgoNs(minutes || 30);
    const data = await lokiQuery(query, start, nowNs(), limit || 500);
    const lines = flattenStreams(data);
    res.json({ namespace, count: lines.length, lines });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// Error/warn log search
app.post('/api/errors', async (req, res) => {
  try {
    const { namespace, pod, minutes, limit, level } = req.body;
    const lvl = level || 'error|warn|ERROR|WARN|Error|Warning|FATAL|fatal|panic|PANIC';

    let selector = '{';
    if (namespace) selector += `namespace="${namespace}",`;
    if (pod) selector += `pod=~"${pod}.*",`;
    selector = selector.replace(/,$/, '') + '}';
    if (selector === '}') selector = '{job=~".+"}';

    const query = `${selector} |~ "${lvl}"`;
    const start = minutesAgoNs(minutes || 60);
    const data = await lokiQuery(query, start, nowNs(), limit || 200);
    const lines = flattenStreams(data);
    res.json({ count: lines.length, filter: lvl, lines });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// Full-text search
app.post('/api/search', async (req, res) => {
  try {
    const { text, namespace, pod, minutes, limit } = req.body;
    if (!text) return res.status(400).json({ error: 'text is required' });

    let selector = '{';
    if (namespace) selector += `namespace="${namespace}",`;
    if (pod) selector += `pod=~"${pod}.*",`;
    selector = selector.replace(/,$/, '') + '}';
    if (selector === '}') selector = '{job=~".+"}';

    const query = `${selector} |~ "${text}"`;
    const start = minutesAgoNs(minutes || 60);
    const data = await lokiQuery(query, start, nowNs(), limit || 200);
    const lines = flattenStreams(data);
    res.json({ count: lines.length, search: text, lines });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// OOMKilled events
app.post('/api/oomkilled', async (req, res) => {
  try {
    const { hours, limit } = req.body;
    const query = `{job=~".+"} |~ "OOMKill|OOM|out of memory|Killed|oomkilled"`;
    const start = hoursAgoNs(hours || 24);
    const data = await lokiQuery(query, start, nowNs(), limit || 100);
    const lines = flattenStreams(data);
    res.json({ count: lines.length, period_hours: hours || 24, lines });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// CrashLoopBackOff events
app.post('/api/crash_loops', async (req, res) => {
  try {
    const { hours, limit } = req.body;
    const query = `{job=~".+"} |~ "CrashLoopBackOff|BackOff|back-off|restarting failed"`;
    const start = hoursAgoNs(hours || 24);
    const data = await lokiQuery(query, start, nowNs(), limit || 100);
    const lines = flattenStreams(data);
    res.json({ count: lines.length, period_hours: hours || 24, lines });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// Log ingestion rate by namespace
app.post('/api/rate', async (req, res) => {
  try {
    const { minutes } = req.body;
    const dur = `${minutes || 5}m`;
    const query = `sum by (namespace) (rate({namespace=~".+"}[${dur}]))`;
    const data = await lokiInstant(query);
    const results = (data.data?.result || [])
      .map(r => ({ namespace: r.metric.namespace, lines_per_second: parseFloat(r.value[1]).toFixed(4) }))
      .sort((a, b) => b.lines_per_second - a.lines_per_second);
    res.json({ count: results.length, window: dur, results });
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

// Loki server stats
app.post('/api/stats', async (req, res) => {
  try {
    const r = await axios.get(`${LOKI_URL}/loki/api/v1/status/buildinfo`);
    res.json(r.data);
  } catch (e) {
    res.status(500).json({ error: e.message });
  }
});

app.use((req, res) => res.status(404).json({ error: 'Not found' }));

app.listen(port, '0.0.0.0', () => {
  console.log(`╔═══════════════════════════════════════════════════════╗`);
  console.log(`║     MCP Loki Server                                   ║`);
  console.log(`║     Port: ${port}                                         ║`);
  console.log(`║     Loki: ${LOKI_URL.substring(0, 42).padEnd(42)} ║`);
  console.log(`╚═══════════════════════════════════════════════════════╝`);
});