keyhan
22359be40e
Close billing, tenancy, migration, build, and CI/CD gaps identified in the audit: wallet/gateway guards, full-UUID namespaces, idempotent migrations with base schema, stateful service stability, safer Dockerfiles/git builds, and platform chart hardening (Redis auth, RollingUpdate, backups, Swagger off). Co-authored-by: Cursor <cursoragent@cursor.com>
989 lines
32 KiB
TypeScript
989 lines
32 KiB
TypeScript
import { Injectable, Logger, ServiceUnavailableException, Inject, forwardRef, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
|
|
import { ConfigService } from '@nestjs/config';
|
|
import * as k8s from '@kubernetes/client-node';
|
|
import * as crypto from 'crypto';
|
|
import { ChildProcess, spawn } from 'child_process';
|
|
import { ClustersService } from '../clusters/clusters.service';
|
|
import { HelmService, LOGGING_HELM_NAMESPACE, LOGGING_HELM_RELEASE } from './helm.service';
|
|
import { userNamespace } from './k8s-workload.util';
|
|
|
|
interface ElasticsearchCredentials {
|
|
username: string;
|
|
password: string;
|
|
}
|
|
|
|
export interface LogSearchFilters {
|
|
applicationId?: string;
|
|
applicationName?: string;
|
|
workload?: string;
|
|
level?: string;
|
|
from?: string;
|
|
to?: string;
|
|
search?: string;
|
|
page?: number;
|
|
limit?: number;
|
|
}
|
|
|
|
export interface NormalizedLogEntry {
|
|
id: string;
|
|
timestamp: string;
|
|
level: string;
|
|
message: string;
|
|
applicationId?: string;
|
|
applicationName?: string;
|
|
workload?: string;
|
|
namespace?: string;
|
|
pod?: string;
|
|
}
|
|
|
|
export interface LogSearchResult {
|
|
hits: NormalizedLogEntry[];
|
|
total: number;
|
|
page: number;
|
|
limit: number;
|
|
}
|
|
|
|
export interface LogStatsResult {
|
|
total: number;
|
|
errors: number;
|
|
warnings: number;
|
|
byLevel: Record<string, number>;
|
|
byWorkload: Record<string, number>;
|
|
period: string;
|
|
}
|
|
|
|
export type LoggingDeployStatus = 'not_installed' | 'installing' | 'failed' | 'ready';
|
|
|
|
export interface LoggingDeployState {
|
|
status: LoggingDeployStatus;
|
|
helmReleaseStatus?: string | null;
|
|
health: {
|
|
status: string;
|
|
clusterName: string;
|
|
numberOfNodes: number;
|
|
activePrimaryShards: number;
|
|
activeShards: number;
|
|
kibanaReady: boolean;
|
|
} | null;
|
|
}
|
|
|
|
/**
|
|
* Central Elasticsearch management service.
|
|
* Deploys a shared Elasticsearch + Kibana stack in a dedicated namespace
|
|
* that all user apps can send logs to via Fluent Bit sidecars.
|
|
*/
|
|
@Injectable()
|
|
export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
|
private readonly logger = new Logger(ElasticsearchService.name);
|
|
private readonly ES_NAMESPACE = 'logging';
|
|
private readonly ES_NAME = 'elasticsearch';
|
|
private readonly KIBANA_NAME = 'kibana';
|
|
private portForwardChild: ChildProcess | null = null;
|
|
private portForwardStartedByUs = false;
|
|
private portForwardKubeconfigFile: string | null = null;
|
|
private portForwardClusterId: string | null = null;
|
|
private ensureInFlight: Promise<void> | null = null;
|
|
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
|
private healthCheckTimer: ReturnType<typeof setInterval> | null = null;
|
|
private reconnectAttempt = 0;
|
|
|
|
// Default credentials - should be overridden via env in production
|
|
private readonly ELASTIC_PASSWORD: string;
|
|
private readonly FLUENTBIT_PASSWORD: string;
|
|
private readonly KIBANA_SYSTEM_PASSWORD: string;
|
|
|
|
constructor(
|
|
@Inject(forwardRef(() => ClustersService))
|
|
private clustersService: ClustersService,
|
|
private configService: ConfigService,
|
|
private helmService: HelmService,
|
|
) {
|
|
this.ELASTIC_PASSWORD = this.configService.get('elasticsearch.password') || '';
|
|
this.FLUENTBIT_PASSWORD = this.configService.get('elasticsearch.fluentbitPassword') || '';
|
|
this.KIBANA_SYSTEM_PASSWORD = this.configService.get('elasticsearch.kibanaPassword') || '';
|
|
}
|
|
|
|
async onModuleInit(): Promise<void> {
|
|
await this.ensureLocalElasticsearchAccess({ waitForCluster: true });
|
|
if (this.shouldAutoPortForward()) {
|
|
this.healthCheckTimer = setInterval(() => {
|
|
void this.periodicElasticsearchHealthCheck();
|
|
}, 30_000);
|
|
}
|
|
}
|
|
|
|
onModuleDestroy(): void {
|
|
if (this.healthCheckTimer) {
|
|
clearInterval(this.healthCheckTimer);
|
|
this.healthCheckTimer = null;
|
|
}
|
|
if (this.reconnectTimer) {
|
|
clearTimeout(this.reconnectTimer);
|
|
this.reconnectTimer = null;
|
|
}
|
|
this.stopDevPortForward();
|
|
}
|
|
|
|
private isLoopbackHost(host: string): boolean {
|
|
return host === '127.0.0.1' || host === 'localhost' || host === '::1';
|
|
}
|
|
|
|
private isRunningInKubernetes(): boolean {
|
|
return Boolean(process.env.KUBERNETES_SERVICE_HOST);
|
|
}
|
|
|
|
/** Hostnames only resolvable from inside the workload cluster (not from a laptop). */
|
|
private isClusterInternalHost(host: string): boolean {
|
|
const h = host.toLowerCase();
|
|
return h.includes('svc.cluster.local') || h.includes('.cluster.') || h === 'elasticsearch' || h === 'kibana';
|
|
}
|
|
|
|
private configuredElasticsearchHost(): string {
|
|
return this.configService.get<string>('elasticsearch.host') || `${this.ES_NAME}.${this.ES_NAMESPACE}.svc.cluster.local`;
|
|
}
|
|
|
|
/** HTTP target: loopback when we tunnel; cluster DNS when the API pod runs in-cluster. */
|
|
private effectiveElasticsearchHost(): string {
|
|
const configured = this.configuredElasticsearchHost();
|
|
if (this.shouldAutoPortForward() && this.isClusterInternalHost(configured)) {
|
|
return '127.0.0.1';
|
|
}
|
|
return configured;
|
|
}
|
|
|
|
private shouldAutoPortForward(): boolean {
|
|
if (this.configService.get<string>('elasticsearch.autoPortForward') === 'false') {
|
|
return false;
|
|
}
|
|
if (process.env.ELASTICSEARCH_AUTO_PORT_FORWARD === 'false') {
|
|
return false;
|
|
}
|
|
if (this.isRunningInKubernetes()) {
|
|
return false;
|
|
}
|
|
const nodeEnv = process.env.NODE_ENV || 'development';
|
|
if (nodeEnv === 'production') {
|
|
return false;
|
|
}
|
|
const host = this.configuredElasticsearchHost();
|
|
return this.isLoopbackHost(host) || this.isClusterInternalHost(host);
|
|
}
|
|
|
|
private stopDevPortForward(): void {
|
|
if (!this.portForwardChild) {
|
|
return;
|
|
}
|
|
const startedByUs = this.portForwardStartedByUs;
|
|
const child = this.portForwardChild;
|
|
const kubeconfigFile = this.portForwardKubeconfigFile;
|
|
this.portForwardChild = null;
|
|
this.portForwardStartedByUs = false;
|
|
this.portForwardKubeconfigFile = null;
|
|
this.portForwardClusterId = null;
|
|
child.kill('SIGTERM');
|
|
if (kubeconfigFile) {
|
|
this.helmService.removeTempFile(kubeconfigFile);
|
|
}
|
|
if (startedByUs) {
|
|
this.logger.log('Stopped Elasticsearch kubectl port-forward');
|
|
}
|
|
}
|
|
|
|
private schedulePortForwardReconnect(reason: string): void {
|
|
if (!this.shouldAutoPortForward()) {
|
|
return;
|
|
}
|
|
if (this.reconnectTimer) {
|
|
return;
|
|
}
|
|
const delay = Math.min(60_000, 2_000 * Math.pow(2, this.reconnectAttempt));
|
|
this.reconnectAttempt += 1;
|
|
this.logger.warn(`Elasticsearch port-forward lost (${reason}). Reconnecting in ${Math.round(delay / 1000)}s…`);
|
|
this.reconnectTimer = setTimeout(() => {
|
|
this.reconnectTimer = null;
|
|
void this.ensureLocalElasticsearchAccess().then((ok) => {
|
|
if (ok) {
|
|
this.reconnectAttempt = 0;
|
|
}
|
|
});
|
|
}, delay);
|
|
}
|
|
|
|
private async periodicElasticsearchHealthCheck(): Promise<void> {
|
|
if (!this.shouldAutoPortForward()) {
|
|
return;
|
|
}
|
|
const deployed = await this.isDeployed();
|
|
if (!deployed) {
|
|
return;
|
|
}
|
|
if (await this.probeElasticsearch()) {
|
|
this.reconnectAttempt = 0;
|
|
return;
|
|
}
|
|
this.logger.debug('Elasticsearch health check failed; restoring tunnel…');
|
|
await this.ensureLocalElasticsearchAccess();
|
|
}
|
|
|
|
private async waitForLoggingStack(maxWaitMs = 120_000): Promise<boolean> {
|
|
const started = Date.now();
|
|
while (Date.now() - started < maxWaitMs) {
|
|
if (await this.isDeployed()) {
|
|
return true;
|
|
}
|
|
await new Promise((r) => setTimeout(r, 5_000));
|
|
}
|
|
return false;
|
|
}
|
|
|
|
private async probeElasticsearch(timeoutMs = 3000): Promise<boolean> {
|
|
try {
|
|
const conn = this.getConnectionInfo();
|
|
const auth = Buffer.from(`${conn.username}:${conn.password}`).toString('base64');
|
|
const response = await fetch(`http://${conn.host}:${conn.port}/_cluster/health`, {
|
|
headers: { Authorization: `Basic ${auth}` },
|
|
signal: AbortSignal.timeout(timeoutMs),
|
|
});
|
|
return response.ok;
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private async waitForElasticsearch(maxWaitMs = 15_000): Promise<boolean> {
|
|
const started = Date.now();
|
|
while (Date.now() - started < maxWaitMs) {
|
|
if (await this.probeElasticsearch(2000)) {
|
|
return true;
|
|
}
|
|
await new Promise((r) => setTimeout(r, 400));
|
|
}
|
|
return false;
|
|
}
|
|
|
|
private async startDevPortForward(localPort: number, clusterId?: string): Promise<void> {
|
|
const targetClusterId = clusterId || null;
|
|
if (this.portForwardChild && this.portForwardClusterId === targetClusterId) {
|
|
return;
|
|
}
|
|
if (this.portForwardChild) {
|
|
this.stopDevPortForward();
|
|
await new Promise((r) => setTimeout(r, 300));
|
|
}
|
|
|
|
const { cluster } = await this.getK8sClients(clusterId);
|
|
const kubeconfigFile = await this.helmService.createKubeconfigFile(cluster.kubeconfig);
|
|
this.portForwardKubeconfigFile = kubeconfigFile;
|
|
this.portForwardClusterId = targetClusterId;
|
|
|
|
const args = ['--kubeconfig', kubeconfigFile, 'port-forward', '-n', this.ES_NAMESPACE, `svc/${this.ES_NAME}`, `${localPort}:9200`];
|
|
this.logger.log(`Starting kubectl port-forward to Elasticsearch on cluster "${cluster.name}" (local log search)`);
|
|
const child = spawn('kubectl', args, { stdio: ['ignore', 'pipe', 'pipe'] });
|
|
this.portForwardChild = child;
|
|
this.portForwardStartedByUs = true;
|
|
child.on('exit', (code, signal) => {
|
|
const wasOurs = this.portForwardChild === child;
|
|
if (wasOurs) {
|
|
this.portForwardChild = null;
|
|
this.portForwardStartedByUs = false;
|
|
}
|
|
if (wasOurs) {
|
|
const reason = code !== 0 && code !== null ? `exit code ${code}` : signal ? `signal ${signal}` : 'connection closed';
|
|
this.schedulePortForwardReconnect(reason);
|
|
}
|
|
});
|
|
child.stderr?.on('data', (chunk: Buffer) => {
|
|
const line = chunk.toString().trim();
|
|
if (line && !line.includes('Handling connection')) {
|
|
this.logger.debug(`kubectl port-forward: ${line}`);
|
|
}
|
|
});
|
|
}
|
|
|
|
/**
|
|
* When the API runs on the host with ELASTICSEARCH_HOST=127.0.0.1, open a tunnel to the cluster.
|
|
* Safe to call repeatedly (e.g. after cluster/API restart or port-forward drop).
|
|
*/
|
|
private async ensureLocalElasticsearchAccess(options?: { waitForCluster?: boolean; clusterId?: string }): Promise<boolean> {
|
|
if (this.ensureInFlight) {
|
|
await this.ensureInFlight;
|
|
return this.probeElasticsearch();
|
|
}
|
|
|
|
this.ensureInFlight = this.ensureLocalElasticsearchAccessImpl(options);
|
|
try {
|
|
await this.ensureInFlight;
|
|
return this.probeElasticsearch();
|
|
} finally {
|
|
this.ensureInFlight = null;
|
|
}
|
|
}
|
|
|
|
private async ensureLocalElasticsearchAccessImpl(options?: { waitForCluster?: boolean; clusterId?: string }): Promise<void> {
|
|
if (!this.shouldAutoPortForward()) {
|
|
return;
|
|
}
|
|
|
|
if (await this.probeElasticsearch()) {
|
|
this.reconnectAttempt = 0;
|
|
return;
|
|
}
|
|
|
|
const clusterId = options?.clusterId;
|
|
let deployed = await this.isDeployed(clusterId);
|
|
if (!deployed && options?.waitForCluster) {
|
|
this.logger.log('Waiting for logging stack after cluster reconnect…');
|
|
deployed = await this.waitForLoggingStack();
|
|
}
|
|
if (!deployed) {
|
|
return;
|
|
}
|
|
|
|
const port = this.configService.get<number>('elasticsearch.port') || 9200;
|
|
|
|
// Stale tunnel after sleep/reboot: port may be bound but ES unreachable
|
|
if (this.portForwardChild) {
|
|
this.stopDevPortForward();
|
|
await new Promise((r) => setTimeout(r, 300));
|
|
}
|
|
|
|
await this.startDevPortForward(port, clusterId);
|
|
const ready = await this.waitForElasticsearch(90_000);
|
|
if (ready) {
|
|
this.reconnectAttempt = 0;
|
|
this.logger.log(`Elasticsearch reachable at 127.0.0.1:${port}`);
|
|
} else {
|
|
this.stopDevPortForward();
|
|
this.logger.warn(`Could not reach Elasticsearch on 127.0.0.1:${port}. Will retry. Manual: kubectl port-forward -n ${this.ES_NAMESPACE} svc/${this.ES_NAME} ${port}:9200`);
|
|
this.schedulePortForwardReconnect('probe timeout');
|
|
}
|
|
}
|
|
|
|
private localElasticsearchHint(): string {
|
|
const conn = this.getConnectionInfo();
|
|
const configured = this.configuredElasticsearchHost();
|
|
if (this.isLoopbackHost(conn.host)) {
|
|
return (
|
|
`Ensure port ${conn.port} is forwarded to the cluster (the API auto-starts kubectl port-forward in development). ` +
|
|
`Manual: kubectl port-forward -n ${this.ES_NAMESPACE} svc/${this.ES_NAME} ${conn.port}:9200`
|
|
);
|
|
}
|
|
if (this.isClusterInternalHost(configured)) {
|
|
if (this.isRunningInKubernetes()) {
|
|
return 'Ensure the logging stack is deployed on the same cluster as this API pod ' + `(Helm release ${LOGGING_HELM_RELEASE} in namespace ${LOGGING_HELM_NAMESPACE}).`;
|
|
}
|
|
return (
|
|
'Run the API inside the cluster, or restart the API locally so it can auto port-forward Elasticsearch ' +
|
|
`(kubectl port-forward -n ${this.ES_NAMESPACE} svc/${this.ES_NAME} ${conn.port}:9200 using the cluster kubeconfig).`
|
|
);
|
|
}
|
|
return `Ensure Elasticsearch is listening on ${conn.host}:${conn.port}.`;
|
|
}
|
|
|
|
private async getK8sClients(clusterId?: string) {
|
|
const cluster = clusterId ? await this.clustersService.findOne(clusterId) : await this.clustersService.getDefault();
|
|
|
|
const kc = new k8s.KubeConfig();
|
|
kc.loadFromString(cluster.kubeconfig);
|
|
|
|
return {
|
|
coreApi: kc.makeApiClient(k8s.CoreV1Api),
|
|
appsApi: kc.makeApiClient(k8s.AppsV1Api),
|
|
networkingApi: kc.makeApiClient(k8s.NetworkingV1Api),
|
|
cluster,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Check if central Elasticsearch is deployed and healthy in a cluster.
|
|
*/
|
|
async isDeployed(clusterId?: string): Promise<boolean> {
|
|
const state = await this.getDeployState(clusterId);
|
|
return state.status === 'ready';
|
|
}
|
|
|
|
/**
|
|
* Detailed logging stack state (Helm release + pod readiness).
|
|
*/
|
|
async getDeployState(clusterId?: string): Promise<LoggingDeployState> {
|
|
const { cluster } = await this.getK8sClients(clusterId);
|
|
const helmStatus = await this.helmService.status(LOGGING_HELM_RELEASE, LOGGING_HELM_NAMESPACE, cluster.kubeconfig);
|
|
|
|
let hasEsWorkload = false;
|
|
try {
|
|
const kc = new k8s.KubeConfig();
|
|
kc.loadFromString(cluster.kubeconfig);
|
|
const appsApi = kc.makeApiClient(k8s.AppsV1Api);
|
|
await appsApi.readNamespacedStatefulSet({
|
|
name: this.ES_NAME,
|
|
namespace: this.ES_NAMESPACE,
|
|
});
|
|
hasEsWorkload = true;
|
|
} catch {
|
|
hasEsWorkload = false;
|
|
}
|
|
|
|
if (!helmStatus && !hasEsWorkload) {
|
|
return { status: 'not_installed', helmReleaseStatus: null, health: null };
|
|
}
|
|
|
|
const health = await this.getHealth(clusterId);
|
|
const helmReleaseStatus = helmStatus?.status || null;
|
|
|
|
if (health?.status === 'green') {
|
|
return { status: 'ready', helmReleaseStatus, health };
|
|
}
|
|
|
|
if (helmReleaseStatus === 'failed' || helmReleaseStatus === 'pending-install') {
|
|
return { status: 'failed', helmReleaseStatus, health };
|
|
}
|
|
|
|
if (hasEsWorkload || helmReleaseStatus === 'deployed') {
|
|
return { status: 'installing', helmReleaseStatus, health };
|
|
}
|
|
|
|
return { status: 'installing', helmReleaseStatus, health };
|
|
}
|
|
|
|
private resolveLoggingStorageClass(): string | undefined {
|
|
const storageClass = this.configService.get<string>('platform.storageClass');
|
|
return storageClass?.trim() || undefined;
|
|
}
|
|
|
|
/**
|
|
* Ensure platform StorageClass exists before Helm creates PVCs (same as app deploy chart).
|
|
*/
|
|
private async ensureClusterStorageClass(kubeconfig: string): Promise<void> {
|
|
const storageClass = this.resolveLoggingStorageClass();
|
|
const createStorageClass = this.configService.get<boolean>('platform.createStorageClass') === true;
|
|
if (!storageClass || !createStorageClass) {
|
|
return;
|
|
}
|
|
|
|
const kc = new k8s.KubeConfig();
|
|
kc.loadFromString(kubeconfig);
|
|
const storageApi = kc.makeApiClient(k8s.StorageV1Api);
|
|
const provisioner = this.configService.get<string>('platform.storageProvisioner') || 'rancher.io/local-path';
|
|
|
|
try {
|
|
await storageApi.readStorageClass({ name: storageClass });
|
|
return;
|
|
} catch (err: any) {
|
|
if (err.code !== 404 && err.body?.code !== 404) {
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
await storageApi.createStorageClass({
|
|
body: {
|
|
apiVersion: 'storage.k8s.io/v1',
|
|
kind: 'StorageClass',
|
|
metadata: { name: storageClass },
|
|
provisioner,
|
|
allowVolumeExpansion: true,
|
|
reclaimPolicy: 'Delete',
|
|
volumeBindingMode: 'WaitForFirstConsumer',
|
|
},
|
|
});
|
|
this.logger.log(`Created StorageClass "${storageClass}" (provisioner: ${provisioner})`);
|
|
}
|
|
|
|
private buildLoggingHelmValues(): {
|
|
elasticPassword: string;
|
|
fluentbitPassword: string;
|
|
kibanaSystemPassword: string;
|
|
storageClass?: string;
|
|
images: {
|
|
elasticsearch?: string;
|
|
kibana?: string;
|
|
busybox?: string;
|
|
curl?: string;
|
|
};
|
|
} {
|
|
const storageClass = this.resolveLoggingStorageClass();
|
|
return {
|
|
elasticPassword: this.ELASTIC_PASSWORD,
|
|
fluentbitPassword: this.FLUENTBIT_PASSWORD,
|
|
kibanaSystemPassword: this.KIBANA_SYSTEM_PASSWORD,
|
|
...(storageClass ? { storageClass } : {}),
|
|
images: {
|
|
elasticsearch: this.configService.get<string>('elasticsearch.images.elasticsearch'),
|
|
kibana: this.configService.get<string>('elasticsearch.images.kibana'),
|
|
busybox: this.configService.get<string>('elasticsearch.images.busybox'),
|
|
curl: this.configService.get<string>('elasticsearch.images.curl'),
|
|
},
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get Elasticsearch health and status
|
|
*/
|
|
async getHealth(clusterId?: string): Promise<{
|
|
status: string;
|
|
clusterName: string;
|
|
numberOfNodes: number;
|
|
activePrimaryShards: number;
|
|
activeShards: number;
|
|
kibanaReady: boolean;
|
|
} | null> {
|
|
const { coreApi } = await this.getK8sClients(clusterId);
|
|
|
|
try {
|
|
// Check ES pods
|
|
const esPods = await coreApi.listNamespacedPod({
|
|
namespace: this.ES_NAMESPACE,
|
|
labelSelector: 'app=elasticsearch',
|
|
});
|
|
|
|
const kibanaPods = await coreApi.listNamespacedPod({
|
|
namespace: this.ES_NAMESPACE,
|
|
labelSelector: 'app=kibana',
|
|
});
|
|
|
|
if (esPods.items.length === 0) {
|
|
return null;
|
|
}
|
|
|
|
const esPod = esPods.items[0];
|
|
const isEsReady = esPod.status?.conditions?.some((c) => c.type === 'Ready' && c.status === 'True');
|
|
|
|
const kibanaPod = kibanaPods.items[0];
|
|
const isKibanaReady = kibanaPod?.status?.conditions?.some((c) => c.type === 'Ready' && c.status === 'True') || false;
|
|
|
|
return {
|
|
status: isEsReady ? 'green' : 'yellow',
|
|
clusterName: 'cloudhost-logs',
|
|
numberOfNodes: 1,
|
|
activePrimaryShards: isEsReady ? 1 : 0,
|
|
activeShards: isEsReady ? 1 : 0,
|
|
kibanaReady: isKibanaReady,
|
|
};
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Deploy or repair central Elasticsearch + Kibana stack via Helm.
|
|
* Always runs upgrade --install; does not wait for pods (avoids timeout errors on slow image pulls).
|
|
*/
|
|
async deploy(clusterId?: string): Promise<{ esPassword: string; kibanaUrl: string; deploying: boolean }> {
|
|
const { cluster } = await this.getK8sClients(clusterId);
|
|
|
|
await this.ensureClusterStorageClass(cluster.kubeconfig);
|
|
|
|
const helmValues = this.buildLoggingHelmValues();
|
|
|
|
try {
|
|
await this.helmService.installLoggingStack(cluster.kubeconfig, helmValues, {
|
|
wait: false,
|
|
timeout: '10m',
|
|
});
|
|
} catch (error: any) {
|
|
const release = await this.helmService.status(LOGGING_HELM_RELEASE, LOGGING_HELM_NAMESPACE, cluster.kubeconfig);
|
|
if (!release) {
|
|
throw error;
|
|
}
|
|
this.logger.warn(`Helm logging install reported an error but release exists (${release.status}); continuing: ${error.message}`);
|
|
}
|
|
|
|
this.logger.log(`Central logging stack applied via Helm (${LOGGING_HELM_RELEASE} in ${LOGGING_HELM_NAMESPACE})`);
|
|
|
|
const state = await this.getDeployState(clusterId);
|
|
|
|
return {
|
|
esPassword: this.ELASTIC_PASSWORD,
|
|
kibanaUrl: `http://kibana.${this.ES_NAMESPACE}.svc.cluster.local:5601`,
|
|
deploying: state.status !== 'ready',
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Undeploy Elasticsearch stack (Helm release; PVC retained by chart policy).
|
|
*/
|
|
async undeploy(clusterId?: string): Promise<void> {
|
|
const { cluster } = await this.getK8sClients(clusterId);
|
|
|
|
try {
|
|
await this.helmService.uninstall(LOGGING_HELM_RELEASE, LOGGING_HELM_NAMESPACE, cluster.kubeconfig);
|
|
this.logger.log('Elasticsearch stack undeployed via Helm (PVC preserved)');
|
|
} catch (e: any) {
|
|
const msg = e?.message || String(e);
|
|
if (msg.includes('not found')) {
|
|
this.logger.log('Logging Helm release not found — nothing to undeploy');
|
|
return;
|
|
}
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get Elasticsearch connection info for apps
|
|
*/
|
|
getConnectionInfo(): {
|
|
host: string;
|
|
port: number;
|
|
username: string;
|
|
password: string;
|
|
} {
|
|
return {
|
|
host: this.effectiveElasticsearchHost(),
|
|
port: this.configService.get<number>('elasticsearch.port') || 9200,
|
|
username: 'elastic',
|
|
password: this.ELASTIC_PASSWORD,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get Fluent Bit credentials for log shipping
|
|
*/
|
|
getFluentBitCredentials(): ElasticsearchCredentials {
|
|
return {
|
|
username: 'elastic',
|
|
password: this.ELASTIC_PASSWORD,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Generate a unique user for log access
|
|
*/
|
|
generateUserCredentials(userId: string): ElasticsearchCredentials {
|
|
const hash = crypto.createHash('sha256').update(`${userId}-${this.ELASTIC_PASSWORD}`).digest('hex');
|
|
return {
|
|
username: userNamespace(userId),
|
|
password: hash.substring(0, 24),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get logs URL for an application (Kibana discover with filter)
|
|
*/
|
|
getLogsUrl(appName: string, namespace: string, kibanaHost: string): string {
|
|
const filter = encodeURIComponent(`app:${appName} AND namespace:${namespace}`);
|
|
return `https://${kibanaHost}/app/discover#/?_g=(filters:!(),query:(language:kuery,query:'${filter}'))`;
|
|
}
|
|
|
|
/**
|
|
* Get index pattern for a user's applications
|
|
*/
|
|
getIndexPattern(userId: string): string {
|
|
return `logs-${userNamespace(userId)}-*`;
|
|
}
|
|
|
|
/**
|
|
* Build must clauses for user log isolation (new + legacy fields).
|
|
*/
|
|
buildUserLogMustClauses(userId: string, filters: LogSearchFilters = {}): any[] {
|
|
// Full-UUID namespace — a truncated prefix would match other tenants'
|
|
// namespaces and leak their logs.
|
|
const namespace = userNamespace(userId);
|
|
|
|
const must: any[] = [
|
|
{
|
|
bool: {
|
|
should: [{ term: { ownerId: userId } }, { term: { 'ownerId.keyword': userId } }, { term: { namespace } }, { term: { 'namespace.keyword': namespace } }],
|
|
minimum_should_match: 1,
|
|
},
|
|
},
|
|
];
|
|
|
|
if (filters.applicationId) {
|
|
must.push({
|
|
bool: {
|
|
should: [{ term: { applicationId: filters.applicationId } }, { term: { 'applicationId.keyword': filters.applicationId } }],
|
|
minimum_should_match: 1,
|
|
},
|
|
});
|
|
}
|
|
|
|
if (filters.applicationName) {
|
|
must.push({
|
|
bool: {
|
|
should: [
|
|
{ term: { applicationName: filters.applicationName } },
|
|
{ term: { 'applicationName.keyword': filters.applicationName } },
|
|
{ term: { app: filters.applicationName } },
|
|
{ term: { 'app.keyword': filters.applicationName } },
|
|
],
|
|
minimum_should_match: 1,
|
|
},
|
|
});
|
|
}
|
|
|
|
if (filters.workload) {
|
|
must.push({
|
|
bool: {
|
|
should: [{ term: { workload: filters.workload } }, { term: { 'workload.keyword': filters.workload } }],
|
|
minimum_should_match: 1,
|
|
},
|
|
});
|
|
}
|
|
|
|
if (filters.level) {
|
|
must.push({
|
|
bool: {
|
|
should: [{ term: { level: filters.level.toLowerCase() } }, { term: { 'level.keyword': filters.level.toLowerCase() } }],
|
|
minimum_should_match: 1,
|
|
},
|
|
});
|
|
}
|
|
|
|
if (filters.from || filters.to) {
|
|
const rangeFilter: any = { range: { '@timestamp': {} } };
|
|
if (filters.from) rangeFilter.range['@timestamp'].gte = filters.from;
|
|
if (filters.to) rangeFilter.range['@timestamp'].lte = filters.to;
|
|
must.push(rangeFilter);
|
|
}
|
|
|
|
if (filters.search) {
|
|
must.push({
|
|
multi_match: {
|
|
query: filters.search,
|
|
fields: ['message', 'log', 'msg', 'error.message'],
|
|
type: 'phrase_prefix',
|
|
},
|
|
});
|
|
}
|
|
|
|
return must;
|
|
}
|
|
|
|
getUserLogsQuery(userId: string): { query: { bool: { must: any[] } } } {
|
|
return {
|
|
query: {
|
|
bool: {
|
|
must: this.buildUserLogMustClauses(userId),
|
|
},
|
|
},
|
|
};
|
|
}
|
|
|
|
getUserIndexPattern(userId: string): string {
|
|
return `logs-${userNamespace(userId)}-*`;
|
|
}
|
|
|
|
private elasticsearchFetch(url: string, auth: string, body: unknown): Promise<Response> {
|
|
return fetch(url, {
|
|
method: body === undefined ? 'GET' : 'POST',
|
|
headers: {
|
|
'Content-Type': 'application/json',
|
|
Authorization: `Basic ${auth}`,
|
|
},
|
|
body: body === undefined ? undefined : JSON.stringify(body),
|
|
signal: AbortSignal.timeout(15_000),
|
|
});
|
|
}
|
|
|
|
private async esRequest(path: string, body: unknown, clusterId?: string): Promise<any> {
|
|
const deployed = await this.isDeployed(clusterId);
|
|
if (!deployed) {
|
|
throw new ServiceUnavailableException('Central logging is not configured. Ask an administrator to deploy Elasticsearch.');
|
|
}
|
|
|
|
if (this.shouldAutoPortForward()) {
|
|
await this.ensureLocalElasticsearchAccess({ clusterId });
|
|
}
|
|
|
|
const conn = this.getConnectionInfo();
|
|
const url = `http://${conn.host}:${conn.port}${path}`;
|
|
const auth = Buffer.from(`${conn.username}:${conn.password}`).toString('base64');
|
|
|
|
let response: Response | undefined;
|
|
try {
|
|
response = await this.elasticsearchFetch(url, auth, body);
|
|
} catch (err: any) {
|
|
if (this.shouldAutoPortForward()) {
|
|
await this.ensureLocalElasticsearchAccess({ clusterId });
|
|
try {
|
|
response = await this.elasticsearchFetch(url, auth, body);
|
|
} catch {
|
|
// retry failed
|
|
}
|
|
}
|
|
if (!response) {
|
|
this.logger.warn(`Elasticsearch unreachable at ${conn.host}:${conn.port}: ${err?.message || err}`);
|
|
throw new ServiceUnavailableException(`Cannot reach Elasticsearch. ${this.localElasticsearchHint()}`);
|
|
}
|
|
}
|
|
|
|
if (!response.ok) {
|
|
const text = await response.text();
|
|
this.logger.warn(`Elasticsearch request failed: ${response.status} ${text}`);
|
|
throw new ServiceUnavailableException('Failed to query log storage');
|
|
}
|
|
|
|
return response.json();
|
|
}
|
|
|
|
private normalizeHit(hit: any): NormalizedLogEntry {
|
|
const src = hit._source || {};
|
|
const message = src.message || src.log || src.msg || (typeof src.error === 'string' ? src.error : src.error?.message) || '';
|
|
|
|
return {
|
|
id: hit._id || '',
|
|
timestamp: src['@timestamp'] || src.timestamp || new Date().toISOString(),
|
|
level: (src.level || 'info').toString().toLowerCase(),
|
|
message: typeof message === 'string' ? message : JSON.stringify(message),
|
|
applicationId: src.applicationId,
|
|
applicationName: src.applicationName || src.app,
|
|
workload: src.workload || 'app',
|
|
namespace: src.namespace,
|
|
pod: src.kubernetes?.pod_name || src.pod_name,
|
|
};
|
|
}
|
|
|
|
async searchLogs(userId: string, filters: LogSearchFilters, clusterId?: string): Promise<LogSearchResult> {
|
|
const page = filters.page || 1;
|
|
const limit = Math.min(filters.limit || 100, 1000);
|
|
const from = (page - 1) * limit;
|
|
|
|
const body = {
|
|
query: { bool: { must: this.buildUserLogMustClauses(userId, filters) } },
|
|
sort: [{ '@timestamp': 'desc' }],
|
|
from,
|
|
size: limit,
|
|
};
|
|
|
|
const index = this.getUserIndexPattern(userId);
|
|
const result = await this.esRequest(`/${index}/_search`, body, clusterId);
|
|
const hits = result.hits?.hits || [];
|
|
|
|
return {
|
|
hits: hits.map((h: any) => this.normalizeHit(h)),
|
|
total: result.hits?.total?.value ?? result.hits?.total ?? hits.length,
|
|
page,
|
|
limit,
|
|
};
|
|
}
|
|
|
|
async searchLogStats(
|
|
userId: string,
|
|
filters: {
|
|
applicationId?: string;
|
|
applicationName?: string;
|
|
workload?: string;
|
|
period?: string;
|
|
},
|
|
clusterId?: string,
|
|
): Promise<LogStatsResult> {
|
|
const periodMap: Record<string, string> = {
|
|
'1h': 'now-1h',
|
|
'6h': 'now-6h',
|
|
'24h': 'now-24h',
|
|
'7d': 'now-7d',
|
|
};
|
|
const period = filters.period || '24h';
|
|
const timeRange = periodMap[period] || 'now-24h';
|
|
|
|
const must = this.buildUserLogMustClauses(userId, {
|
|
applicationId: filters.applicationId,
|
|
applicationName: filters.applicationName,
|
|
workload: filters.workload,
|
|
});
|
|
must.push({ range: { '@timestamp': { gte: timeRange } } });
|
|
|
|
const body = {
|
|
query: { bool: { must } },
|
|
size: 0,
|
|
aggs: {
|
|
by_level: {
|
|
terms: { field: 'level.keyword', size: 10, missing: 'unknown' },
|
|
},
|
|
by_workload: {
|
|
terms: { field: 'workload.keyword', size: 10, missing: 'app' },
|
|
},
|
|
error_count: { filter: { term: { 'level.keyword': 'error' } } },
|
|
warn_count: { filter: { term: { 'level.keyword': 'warn' } } },
|
|
},
|
|
};
|
|
|
|
const index = this.getUserIndexPattern(userId);
|
|
const result = await this.esRequest(`/${index}/_search`, body, clusterId);
|
|
|
|
const byLevel: Record<string, number> = {};
|
|
for (const bucket of result.aggregations?.by_level?.buckets || []) {
|
|
byLevel[bucket.key] = bucket.doc_count;
|
|
}
|
|
|
|
const byWorkload: Record<string, number> = {};
|
|
for (const bucket of result.aggregations?.by_workload?.buckets || []) {
|
|
byWorkload[bucket.key] = bucket.doc_count;
|
|
}
|
|
|
|
const errors = result.aggregations?.error_count?.doc_count || 0;
|
|
const warnings = result.aggregations?.warn_count?.doc_count || 0;
|
|
const total = Object.values(byLevel).reduce((a, b) => a + b, 0);
|
|
|
|
return { total, errors, warnings, byLevel, byWorkload, period };
|
|
}
|
|
|
|
async searchRecentErrors(
|
|
userId: string,
|
|
filters: {
|
|
applicationId?: string;
|
|
applicationName?: string;
|
|
workload?: string;
|
|
hours?: number;
|
|
limit?: number;
|
|
},
|
|
clusterId?: string,
|
|
): Promise<NormalizedLogEntry[]> {
|
|
const hours = filters.hours || 24;
|
|
const limit = Math.min(filters.limit || 50, 500);
|
|
|
|
const must = this.buildUserLogMustClauses(userId, {
|
|
applicationId: filters.applicationId,
|
|
applicationName: filters.applicationName,
|
|
workload: filters.workload,
|
|
level: 'error',
|
|
});
|
|
must.push({ range: { '@timestamp': { gte: `now-${hours}h` } } });
|
|
|
|
const body = {
|
|
query: { bool: { must } },
|
|
sort: [{ '@timestamp': 'desc' }],
|
|
size: limit,
|
|
};
|
|
|
|
const index = this.getUserIndexPattern(userId);
|
|
const result = await this.esRequest(`/${index}/_search`, body, clusterId);
|
|
return (result.hits?.hits || []).map((h: any) => this.normalizeHit(h));
|
|
}
|
|
|
|
async getLoggingStatus(clusterId?: string): Promise<{
|
|
available: boolean;
|
|
deployed: boolean;
|
|
recovering?: boolean;
|
|
message?: string;
|
|
}> {
|
|
let deployed = await this.isDeployed(clusterId);
|
|
if (!deployed && this.shouldAutoPortForward()) {
|
|
deployed = await this.waitForLoggingStack(8_000);
|
|
}
|
|
if (!deployed) {
|
|
return {
|
|
available: false,
|
|
deployed: false,
|
|
message: 'Central logging is not deployed. Ask an administrator to deploy Elasticsearch.',
|
|
};
|
|
}
|
|
|
|
if (this.shouldAutoPortForward() && !(await this.probeElasticsearch())) {
|
|
void this.ensureLocalElasticsearchAccess({ clusterId });
|
|
}
|
|
|
|
if (await this.probeElasticsearch()) {
|
|
return { available: true, deployed: true };
|
|
}
|
|
|
|
return {
|
|
available: false,
|
|
deployed: true,
|
|
recovering: this.shouldAutoPortForward(),
|
|
message: this.shouldAutoPortForward()
|
|
? 'Reconnecting to Elasticsearch after cluster or API restart. This usually takes under a minute.'
|
|
: `Elasticsearch is running in the cluster, but this backend cannot reach it. ${this.localElasticsearchHint()}`,
|
|
};
|
|
}
|
|
}
|