chore: add prometheus, loki, and vector e2e testing

.codespellrc

@@ -1,6 +1,6 @@
 # Lint Codespell configurations
 [codespell]
 skip = .codespellrc,.git,node_modules,build,dist,*.zst,CHANGELOG.md,.playwright,.terraform
-ignore-words-list = NotIn,AKS,LICENS,aks
+ignore-words-list = NotIn,AKS,LICENS,aks,afterAll
 enable-colors =
 check-hidden =

.yamllint

@@ -7,7 +7,7 @@ ignore:
   - 'k3d/local/manifests/metallb/metallb-native.yaml'
   - '**/.terraform/**'
   - '**/chart/templates/**'
-  - 'node_modules/**'
+  - '**/node_modules/**'
   - 'dist/**'
   - 'src/pepr/uds-operator-config/templates/**'
   - '.codespellrc'

e2e/test/forward.ts

@@ -0,0 +1,117 @@
+/**
+ * Copyright 2024 Defense Unicorns
+ * SPDX-License-Identifier: AGPL-3.0-or-later OR LicenseRef-Defense-Unicorns-Commercial
+ */
+
+import * as k8s from '@kubernetes/client-node';
+import { K8s, kind } from 'kubernetes-fluent-client';
+import * as net from 'net';
+
+const kc = new k8s.KubeConfig();
+const forward = new k8s.PortForward(kc);
+kc.loadFromDefault();
+
+interface ForwardResult {
+  server: net.Server;
+  url: string;
+}
+
+// Utility function to get an available random port within a range
+async function getAvailablePort(min = 1024, max = 65535): Promise<number> {
+  let port: number;
+  let isAvailable = false;
+
+  while (!isAvailable) {
+    port = Math.floor(Math.random() * (max - min + 1)) + min;
+    isAvailable = await new Promise<boolean>((resolve) => {
+      const server = net.createServer();
+
+      server.once('error', () => resolve(false)); // Port is in use
+      server.once('listening', () => {
+        server.close(() => resolve(true)); // Port is available
+      });
+
+      server.listen(port, '127.0.0.1');
+    });
+  }
+
+  return port!;
+}
+
+export async function getPodFromService(svc: string, namespace: string): Promise<string> {
+  try {
+    const service = await K8s(kind.Service).InNamespace(namespace).Get(svc);
+    const labelSelector = service.spec?.selector;
+
+    if (!labelSelector) {
+      throw new Error(`No label selectors found for service: ${svc}`);
+    }
+
+    let podsQuery = K8s(kind.Pod).InNamespace(namespace);
+    for (const key in labelSelector) {
+      podsQuery = podsQuery.WithLabel(key, labelSelector[key]);
+    }
+
+    const pods = await podsQuery.Get();
+    if (pods.items.length === 0) {
+      throw new Error(`No pods found for service: ${svc}`);
+    }
+
+    return pods.items[0].metadata!.name!;
+  } catch (err) {
+    // Type guard to check if `err` is an instance of `Error`
+    if (err instanceof Error) {
+      throw new Error(`Failed to get pod from service ${svc}: ${err.message}`);
+    } else {
+      throw new Error(`Unknown error occurred while fetching pod from service ${svc}`);
+    }
+  }
+}
+
+export async function getForward(service: string, namespace: string, port: number): Promise<ForwardResult> {
+  try {
+    const podName = await getPodFromService(service, namespace);
+    const randomPort = await getAvailablePort(3000, 65535);
+
+    return await new Promise<ForwardResult>((resolve, reject) => {
+      const server = net.createServer((socket) => {
+        forward.portForward(namespace, podName, [port], socket, null, socket);
+      });
+
+      server.listen(randomPort, '127.0.0.1', () => {
+        resolve({ server, url: `http://localhost:${randomPort}` });
+      });
+
+      server.on('error', (err) => {
+        // Type guard to check if `err` is an instance of `Error`
+        if (err instanceof Error) {
+          reject(new Error(`Error binding to port ${randomPort}: ${err.message}`));
+        } else {
+          reject(new Error(`Unknown error occurred while binding to port ${randomPort}`));
+        }
+      });
+    });
+  } catch (err) {
+    // Type guard to check if `err` is an instance of `Error`
+    if (err instanceof Error) {
+      throw new Error(`Failed to setup port forwarding for service ${service}: ${err.message}`);
+    } else {
+      throw new Error(`Unknown error occurred while setting up port forwarding for service ${service}`);
+    }
+  }
+}
+
+export function closeForward(server: net.Server): Promise<void> {
+  return new Promise((resolve, reject) => {
+    server.close((err) => {
+      // Type guard to check if `err` is an instance of `Error`
+      if (err instanceof Error) {
+        reject(new Error(`Failed to close server: ${err.message}`));
+      } else if (err) {
+        reject(new Error('Unknown error occurred while closing the server'));
+      } else {
+        resolve();
+      }
+    });
+  });
+}

e2e/test/loki.spec.ts

@@ -0,0 +1,182 @@
+/**
+ * Copyright 2024 Defense Unicorns
+ * SPDX-License-Identifier: AGPL-3.0-or-later OR LicenseRef-Defense-Unicorns-Commercial
+ */
+
+import { afterAll, beforeAll, describe, expect, test } from "@jest/globals";
+import * as net from 'net';
+import { closeForward, getForward } from './forward';
+
+// Global variables
+let lokiBackend: { server: net.Server, url: string };
+let lokiRead: { server: net.Server, url: string };
+let lokiWrite: { server: net.Server, url: string };
+let lokiGateway: { server: net.Server, url: string };
+
+// Helper functions
+const getLokiUrl = (path: string, component: { url: string }) => `${component.url}${path}`;
+
+// Reuse Loki URL for sending logs
+const sendLog = async (
+    logMessage: string,
+    labels: Record<string, string>,
+    timestamp: string = `${Date.now() * 1_000_000}`,
+    expectReject: boolean = false
+): Promise<void> => {
+    const logEntry = {
+        streams: [{ stream: labels, values: [[timestamp, logMessage]] }],
+    };
+
+    try {
+        const response = await fetch(getLokiUrl('/loki/api/v1/push', lokiWrite), {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify(logEntry),
+        });
+
+        if (!response.ok) {
+            if (expectReject && response.status === 400) return;
+            throw new Error(`Unexpected log ingestion failure: ${logMessage}`);
+        }
+
+        if (expectReject) throw new Error(`Unexpected log acceptance: ${logMessage}`);
+    } catch (error: any) {
+        if (!(expectReject && error.message.includes('400'))) {
+            console.error(`Error in log ingestion: ${logMessage}`, error.message);
+            throw error;
+        }
+    }
+};
+
+const queryLogs = async (query: string, limit = 1): Promise<any> => {
+    try {
+        const response = await fetch(getLokiUrl(`/loki/api/v1/query_range?query=${encodeURIComponent(query)}&limit=${limit}`, lokiRead), {
+            method: 'GET',
+        });
+        if (!response.ok) throw new Error('Error in querying logs');
+        return response.json();
+    } catch (error: any) {
+        console.error('Error in querying logs', error.message);
+        throw error;
+    }
+};
+
+const checkLokiServices = async (component: string, expectedServices: string[], urlComponent: { url: string }): Promise<void> => {
+    try {
+        const response = await fetch(getLokiUrl('/services', urlComponent));
+        if (!response.ok) throw new Error(`Error checking services for ${component}`);
+
+        const servicesList = (await response.text()).split('\n');
+        expectedServices.forEach(service => {
+            if (!servicesList.some(svc => svc.includes(`${service} => Running`))) {
+                throw new Error(`${service} is not running for ${component}`);
+            }
+        });
+    } catch (error: any) {
+        console.error(`Error checking services for ${component}`, error.message);
+        throw error;
+    }
+};
+
+// Unified log validation function
+const validateLogInQuery = (queryData: any, logMessage: string): void => {
+    expect(queryData).toHaveProperty('status', 'success');
+    expect(Array.isArray(queryData.data.result)).toBe(true);
+    const logExists = queryData.data.result.some((stream: any) =>
+        stream.values.some((value: any) => value.includes(logMessage))
+    );
+    expect(logExists).toBe(true);
+};
+
+// Jest test cases
+describe('Loki Tests', () => {
+    beforeAll(async () => {
+        lokiBackend = await getForward('loki-backend', 'loki', 3100);
+        lokiRead = await getForward('loki-read', 'loki', 3100);
+        lokiWrite = await getForward('loki-write', 'loki', 3100);
+        lokiGateway = await getForward('loki-gateway', 'loki', 8080);
+    });
+
+    afterAll(async () => {
+        await closeForward(lokiBackend.server);
+        await closeForward(lokiRead.server);
+        await closeForward(lokiWrite.server);
+        await closeForward(lokiGateway.server);
+    });
+
+    test('Validate Vector logs are present in Loki (loki-read)', async () => {
+        const data = await queryLogs('{collector="vector"}');
+        expect(data).toHaveProperty('status', 'success');
+        expect(Array.isArray(data.data.result)).toBe(true);
+    });
+
+    test('Validate node logs from vector are present in Loki', async () => {
+        const data = await queryLogs('{service_name="varlogs", collector="vector"}');
+        expect(data).toHaveProperty('status', 'success');
+        expect(Array.isArray(data.data.result)).toBe(true);
+    });
+
+
+    test('Validate pod logs from vector are present in Loki', async () => {
+        const data = await queryLogs('{service_name="pepr-uds-core", collector="vector"}');
+        expect(data).toHaveProperty('status', 'success');
+        expect(Array.isArray(data.data.result)).toBe(true);
+    });
+
+    test('Send log to Loki-write and validate in Loki-read', async () => {
+        const logMessage = 'Test log from jest';
+        await sendLog(logMessage, { job: 'test-job', level: 'info' });
+        const data = await queryLogs('{job="test-job"}');
+        validateLogInQuery(data, logMessage);
+    });
+
+    test('Check services are running for loki-read', async () => {
+        const expectedServices = [
+            'querier', 'server', 'runtime-config', 'ring', 'query-scheduler-ring',
+            'memberlist-kv', 'cache-generation-loader', 'ingester-querier'
+        ];
+        await checkLokiServices('loki-read', expectedServices, lokiRead);
+    });
+
+    test('Check services are running for loki-write', async () => {
+        const expectedServices = ['ring', 'store', 'ingester', 'distributor', 'runtime-config', 'server', 'memberlist-kv'];
+        await checkLokiServices('loki-write', expectedServices, lokiWrite);
+    });
+
+    test('Check services are running for loki-backend', async () => {
+        const expectedServices = [
+            'compactor', 'index-gateway', 'ring', 'query-scheduler-ring',
+            'index-gateway-ring', 'ingester-querier', 'store', 'server',
+            'memberlist-kv', 'runtime-config', 'query-scheduler', 'ruler'
+        ];
+        await checkLokiServices('loki-backend', expectedServices, lokiBackend);
+    });
+
+    test('Validate Loki Gateway is responsive', async () => {
+        const response = await fetch(`${lokiGateway.url}`);
+        expect(response.status).toBe(200);
+    });
+
+    test('Send log to Loki-gateway and validate it in Loki-read and Loki-write', async () => {
+        const logMessage = 'Test log via gateway';
+        const labels = { job: 'gateway-test', level: 'info' };
+
+        // Send log to loki-gateway
+        const logEntry = { streams: [{ stream: labels, values: [[`${Date.now() * 1_000_000}`, logMessage]] }] };
+        const response = await fetch(getLokiUrl('/loki/api/v1/push', lokiGateway), {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify(logEntry),
+        });
+
+        expect(response.ok).toBe(true);
+
+        // Validate the log appears in Loki-read
+        const readData = await queryLogs(`{job="gateway-test"}`);
+        validateLogInQuery(readData, logMessage);
+
+        // Validate the log appears in Loki-write
+        const writeData = await queryLogs(`{job="gateway-test"}`);
+        validateLogInQuery(writeData, logMessage);
+    });
+});