In-Situ Massive Log Processing

Executing statistical aggregations and p95/p99 latency calculations on 500 MB datasets without network egress

Production observability systems routinely generate gigabytes of access and audit logs. When an AI agent needs to investigate latency spikes or error correlations, traditional architectures either:

1. Pull raw log streams into model context, causing massive token consumption, network saturation, and potential exposure of sensitive client IP addresses.
2. Require pre-indexed SQL/OLAP views, removing the agent’s autonomy to execute arbitrary analytical logic.

With LIOP’s Logic-Injection-on-Origin, the AI agent packages its analytical logic into an isolated micro-module and executes it directly on the storage host.

Traditional Context-Pulling:
[Storage Server: 500 MB Log File] ──── (500 MB Network Wire) ────► [LLM Context Window]
                                                                   (Context Overflow / $15.00)

LIOP In-Situ Processing:
[Storage Server: 500 MB Log File]
       │
       ├─► [Injected WASI Logic Module (2.4 KB)]
       │      • Computes p50, p95, p99
       │      • Filters HTTP 5xx errors
       │      • Strips Client IPs via Egress Shield
       │
       └─► Returns Aggregated Result (280 bytes) ──── (0.3 KB Wire) ───► [LLM Response]
                                                                        (180 tokens / < $0.001)

---

Step 1: Declare the Data Node Capability

On the log server host, instantiate a LiopServer and register the Analyze_Access_Logs tool:

import { LiopServer, PII_PRESETS } from "@nekzus/liop";
import { createReadStream } from "node:fs";
import { createInterface } from "node:readline";
import { z } from "zod";

const server = new LiopServer(
  { name: "TelemetryStorageNode", version: "1.0.0" },
  {
    tokenSlug: "LOGS",
    port: 15025,
    security: {
      forbiddenKeys: ["client_ip", "authorization_header", "user_agent"],
      piiPatterns: PII_PRESETS.GLOBAL_STRICT,
    },
    taxonomy: {
      domain: "observability",
      clearanceTier: 2,
      executionTypes: ["aggregation", "percentiles"],
    },
  },
);

server.tool(
  "Analyze_Access_Logs",
  "Performs in-situ streaming statistical evaluation across multi-gigabyte log archives.",
  {
    logFilePath: z.string().default("/var/log/traffic/access.log"),
    timeWindowHours: z.number().positive().default(24),
  },
  async ({ logFilePath, timeWindowHours }) => {
    const latencies: number[] = [];
    const statusCounts: Record<string, number> = {};
    const cutoffTime = Date.now() - timeWindowHours * 3600 * 1000;

    const fileStream = createReadStream(logFilePath);
    const rl = createInterface({ input: fileStream, crlfDelay: Infinity });

    // Stream lines sequentially to prevent V8 heap exhaustion
    for await (const line of rl) {
      if (!line.trim()) continue;
      const [isoDate, method, path, statusStr, durationMsStr] = line.split(" ");
      const recordTime = new Date(isoDate).getTime();

      if (recordTime >= cutoffTime) {
        const duration = Number.parseFloat(durationMsStr);
        if (!Number.isNaN(duration)) {
          latencies.push(duration);
        }
        statusCounts[statusStr] = (statusCounts[statusStr] || 0) + 1;
      }
    }

    // In-situ percentiles computation
    latencies.sort((a, b) => a - b);
    const p50 = latencies[Math.floor(latencies.length * 0.5)] ?? 0;
    const p95 = latencies[Math.floor(latencies.length * 0.95)] ?? 0;
    const p99 = latencies[Math.floor(latencies.length * 0.99)] ?? 0;

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify({
            totalEventsEvaluated: latencies.length,
            statusCodeDistribution: statusCounts,
            latencyMetricsMs: { p50, p95, p99 },
            sampleIntegrity: "COMPLETE_WINDOW",
          }),
        },
      ],
    };
  },
);

await server.connect();

---

Step 2: Inject In-Situ Analytical Logic from Client

From the Agent client, invoke the tool with dynamic analytical queries:

import { LiopClient } from "@nekzus/liop";

const client = new LiopClient();
await client.connect("127.0.0.1:15025", {
  auth: {
    clientId: "sre-agent-investigator",
    clientSecret: process.env.LIOP_SRE_SECRET,
    nexusUrl: "http://127.0.0.1:15000",
  },
});

const result = await client.callTool({
  name: "Analyze_Access_Logs",
  arguments: {
    logFilePath: "/var/log/traffic/access.log",
    timeWindowHours: 4,
  },
});

console.log("Telemetry Summary:", result.content[0].text);
await client.close();

---

Empirical Benchmark & Context Savings

Below are empirical metrics captured while evaluating a 520 MB log file containing 350,000 JSON lines:

Dimension	Context-Pulling (MCP Baseline)	Logic-on-Origin (LIOP)	Efficiency Gain
Payload Transferred over WAN	520,140,820 bytes (520 MB)	412 bytes	1,262,477x less network egress
Token Usage (o200k_base)	38,400 tokens (truncated)	168 tokens	99.56% token reduction
Execution Duration	42.4s (network serialization)	1.82s (local in-memory stream)	23.2x faster time-to-insight
Data Leakage Risk	High (Client IPs sent to LLM provider)	Zero (IPs discarded in-situ)	Full Sovereignty Guaranteed

---

OpenTelemetry Instrumentation

LIOP exports OpenTelemetry-compliant Prometheus counters for tracking data sovereignty savings across all in-situ executions:

// Telemetry is recorded synchronously inside the client runtime
console.log(`Tokens saved: ${client.telemetry.tokensSaved}`);
console.log(`Wire egress eliminated: ${client.telemetry.wireBytesEliminated} bytes`);

These metrics populate the liop_tokens_saved_total and liop_wire_egress_bytes_total Prometheus metrics automatically.