From 5ab7a6081ec772e5c0227e4cc5bbe41c4e608380 Mon Sep 17 00:00:00 2001 From: keyhan Date: Thu, 14 May 2026 12:41:56 +0330 Subject: [PATCH] feat(logging): add ElasticsearchService for cluster deployment Deploy and manage central Elasticsearch + Kibana via K8s API, with health checks, credentials, and per-user log query helpers. Co-authored-by: Cursor --- .../src/kubernetes/elasticsearch.service.ts | 525 ++++++++++++++++++ 1 file changed, 525 insertions(+) create mode 100644 backend/src/kubernetes/elasticsearch.service.ts diff --git a/backend/src/kubernetes/elasticsearch.service.ts b/backend/src/kubernetes/elasticsearch.service.ts new file mode 100644 index 0000000..3fb211a --- /dev/null +++ b/backend/src/kubernetes/elasticsearch.service.ts @@ -0,0 +1,525 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import * as k8s from '@kubernetes/client-node'; +import * as crypto from 'crypto'; +import { ClustersService } from '../clusters/clusters.service'; + +interface ElasticsearchCredentials { + username: string; + password: string; +} + +interface LogEntry { + timestamp: string; + level: string; + message: string; + app: string; + namespace: string; + [key: string]: any; +} + +/** + * 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 { + private readonly logger = new Logger(ElasticsearchService.name); + private readonly ES_NAMESPACE = 'logging'; + private readonly ES_NAME = 'elasticsearch'; + private readonly KIBANA_NAME = 'kibana'; + + // Default credentials - should be overridden via env in production + private readonly ELASTIC_PASSWORD: string; + private readonly FLUENTBIT_PASSWORD: string; + + constructor( + private clustersService: ClustersService, + private configService: ConfigService, + ) { + this.ELASTIC_PASSWORD = this.configService.get('elasticsearch.password') || 'CloudHost2024!Secure'; + this.FLUENTBIT_PASSWORD = this.configService.get('elasticsearch.fluentbitPassword') || 'FluentBit2024!Writer'; + } + + 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 in a cluster + */ + async isDeployed(clusterId?: string): Promise { + const { appsApi } = await this.getK8sClients(clusterId); + + try { + await appsApi.readNamespacedStatefulSet(this.ES_NAME, this.ES_NAMESPACE); + return true; + } catch { + return false; + } + } + + /** + * 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( + this.ES_NAMESPACE, + undefined, + undefined, + undefined, + undefined, + 'app=elasticsearch', + ); + + const kibanaPods = await coreApi.listNamespacedPod( + this.ES_NAMESPACE, + undefined, + undefined, + undefined, + undefined, + 'app=kibana', + ); + + if (esPods.body.items.length === 0) { + return null; + } + + const esPod = esPods.body.items[0]; + const isEsReady = esPod.status?.conditions?.some( + (c) => c.type === 'Ready' && c.status === 'True', + ); + + const kibanaPod = kibanaPods.body.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 central Elasticsearch + Kibana stack + * This should be called once per cluster by admin + */ + async deploy(clusterId?: string): Promise<{ esPassword: string; kibanaUrl: string }> { + const { coreApi, appsApi } = await this.getK8sClients(clusterId); + + // 1. Create logging namespace + await this.ensureNamespace(coreApi); + + // 2. Create credentials secret + await this.createCredentialsSecret(coreApi); + + // 3. Deploy Elasticsearch + await this.deployElasticsearch(coreApi, appsApi); + + // 4. Deploy Kibana + await this.deployKibana(coreApi, appsApi); + + this.logger.log('Central Elasticsearch stack deployed successfully'); + + return { + esPassword: this.ELASTIC_PASSWORD, + kibanaUrl: `http://kibana.${this.ES_NAMESPACE}.svc.cluster.local:5601`, + }; + } + + /** + * Undeploy Elasticsearch stack + */ + async undeploy(clusterId?: string): Promise { + const { coreApi, appsApi } = await this.getK8sClients(clusterId); + + try { + // Delete Kibana + await appsApi.deleteNamespacedDeployment(this.KIBANA_NAME, this.ES_NAMESPACE); + await coreApi.deleteNamespacedService(this.KIBANA_NAME, this.ES_NAMESPACE); + this.logger.log('Kibana deleted'); + } catch (e: any) { + if (e?.response?.statusCode !== 404) { + this.logger.warn(`Failed to delete Kibana: ${e.message}`); + } + } + + try { + // Delete Elasticsearch + await appsApi.deleteNamespacedStatefulSet(this.ES_NAME, this.ES_NAMESPACE); + await coreApi.deleteNamespacedService(this.ES_NAME, this.ES_NAMESPACE); + this.logger.log('Elasticsearch deleted'); + } catch (e: any) { + if (e?.response?.statusCode !== 404) { + this.logger.warn(`Failed to delete Elasticsearch: ${e.message}`); + } + } + + // Note: PVC is kept intentionally to preserve data + this.logger.log('Elasticsearch stack undeployed (PVC preserved)'); + } + + private async ensureNamespace(coreApi: k8s.CoreV1Api): Promise { + try { + await coreApi.readNamespace(this.ES_NAMESPACE); + } catch { + await coreApi.createNamespace({ + metadata: { + name: this.ES_NAMESPACE, + labels: { 'app.kubernetes.io/managed-by': 'cloudhost' }, + }, + }); + this.logger.log(`Created namespace: ${this.ES_NAMESPACE}`); + } + } + + private async createCredentialsSecret(coreApi: k8s.CoreV1Api): Promise { + const secret = { + apiVersion: 'v1', + kind: 'Secret', + metadata: { + name: 'elasticsearch-credentials', + namespace: this.ES_NAMESPACE, + }, + type: 'Opaque', + data: { + ELASTIC_PASSWORD: Buffer.from(this.ELASTIC_PASSWORD).toString('base64'), + FLUENTBIT_PASSWORD: Buffer.from(this.FLUENTBIT_PASSWORD).toString('base64'), + }, + }; + + try { + await coreApi.replaceNamespacedSecret('elasticsearch-credentials', this.ES_NAMESPACE, secret); + } catch { + await coreApi.createNamespacedSecret(this.ES_NAMESPACE, secret); + } + } + + private async deployElasticsearch( + coreApi: k8s.CoreV1Api, + appsApi: k8s.AppsV1Api, + ): Promise { + // PVC for Elasticsearch data + const pvc: k8s.V1PersistentVolumeClaim = { + apiVersion: 'v1', + kind: 'PersistentVolumeClaim', + metadata: { name: `${this.ES_NAME}-data`, namespace: this.ES_NAMESPACE }, + spec: { + accessModes: ['ReadWriteOnce'], + resources: { requests: { storage: '50Gi' } }, + }, + }; + + try { + await coreApi.readNamespacedPersistentVolumeClaim(pvc.metadata!.name!, this.ES_NAMESPACE); + } catch { + await coreApi.createNamespacedPersistentVolumeClaim(this.ES_NAMESPACE, pvc); + } + + // Base64 encoded auth header + const authHeader = Buffer.from(`elastic:${this.ELASTIC_PASSWORD}`).toString('base64'); + + // StatefulSet for Elasticsearch + const statefulSet: k8s.V1StatefulSet = { + apiVersion: 'apps/v1', + kind: 'StatefulSet', + metadata: { + name: this.ES_NAME, + namespace: this.ES_NAMESPACE, + labels: { app: this.ES_NAME }, + }, + spec: { + serviceName: this.ES_NAME, + replicas: 1, + selector: { matchLabels: { app: this.ES_NAME } }, + template: { + metadata: { labels: { app: this.ES_NAME } }, + spec: { + securityContext: { fsGroup: 1000 }, + initContainers: [ + { + name: 'fix-permissions', + image: 'busybox:1.36', + command: ['sh', '-c', 'chown -R 1000:1000 /usr/share/elasticsearch/data'], + securityContext: { runAsUser: 0, privileged: true }, + volumeMounts: [{ name: 'es-data', mountPath: '/usr/share/elasticsearch/data' }], + }, + { + name: 'increase-vm-max-map', + image: 'busybox:1.36', + command: ['sysctl', '-w', 'vm.max_map_count=262144'], + securityContext: { privileged: true }, + }, + ], + containers: [ + { + name: 'elasticsearch', + image: 'docker.elastic.co/elasticsearch/elasticsearch:8.12.0', + ports: [ + { containerPort: 9200, name: 'http' }, + { containerPort: 9300, name: 'transport' }, + ], + env: [ + { name: 'discovery.type', value: 'single-node' }, + { name: 'xpack.security.enabled', value: 'true' }, + { name: 'xpack.security.http.ssl.enabled', value: 'false' }, + { name: 'xpack.security.transport.ssl.enabled', value: 'false' }, + { + name: 'ELASTIC_PASSWORD', + valueFrom: { + secretKeyRef: { name: 'elasticsearch-credentials', key: 'ELASTIC_PASSWORD' }, + }, + }, + { name: 'ES_JAVA_OPTS', value: '-Xms1g -Xmx1g' }, + { name: 'cluster.name', value: 'cloudhost-logs' }, + { name: 'bootstrap.memory_lock', value: 'false' }, + ], + resources: { + requests: { cpu: '500m', memory: '2Gi' }, + limits: { cpu: '2000m', memory: '4Gi' }, + }, + volumeMounts: [{ name: 'es-data', mountPath: '/usr/share/elasticsearch/data' }], + readinessProbe: { + httpGet: { + path: '/_cluster/health?local=true', + port: 9200 as any, + httpHeaders: [{ name: 'Authorization', value: `Basic ${authHeader}` }], + }, + initialDelaySeconds: 30, + periodSeconds: 10, + timeoutSeconds: 5, + }, + livenessProbe: { + httpGet: { + path: '/_cluster/health?local=true', + port: 9200 as any, + httpHeaders: [{ name: 'Authorization', value: `Basic ${authHeader}` }], + }, + initialDelaySeconds: 60, + periodSeconds: 30, + timeoutSeconds: 10, + }, + }, + ], + volumes: [ + { + name: 'es-data', + persistentVolumeClaim: { claimName: `${this.ES_NAME}-data` }, + }, + ], + }, + }, + }, + }; + + try { + await appsApi.replaceNamespacedStatefulSet(this.ES_NAME, this.ES_NAMESPACE, statefulSet); + } catch { + await appsApi.createNamespacedStatefulSet(this.ES_NAMESPACE, statefulSet); + } + + // Service for Elasticsearch + const service: k8s.V1Service = { + apiVersion: 'v1', + kind: 'Service', + metadata: { name: this.ES_NAME, namespace: this.ES_NAMESPACE }, + spec: { + selector: { app: this.ES_NAME }, + ports: [ + { port: 9200, targetPort: 9200 as any, name: 'http' }, + { port: 9300, targetPort: 9300 as any, name: 'transport' }, + ], + type: 'ClusterIP', + }, + }; + + try { + await coreApi.replaceNamespacedService(this.ES_NAME, this.ES_NAMESPACE, service); + } catch { + await coreApi.createNamespacedService(this.ES_NAMESPACE, service); + } + + this.logger.log('Elasticsearch deployed'); + } + + private async deployKibana( + coreApi: k8s.CoreV1Api, + appsApi: k8s.AppsV1Api, + ): Promise { + const deployment: k8s.V1Deployment = { + apiVersion: 'apps/v1', + kind: 'Deployment', + metadata: { + name: this.KIBANA_NAME, + namespace: this.ES_NAMESPACE, + labels: { app: this.KIBANA_NAME }, + }, + spec: { + replicas: 1, + selector: { matchLabels: { app: this.KIBANA_NAME } }, + template: { + metadata: { labels: { app: this.KIBANA_NAME } }, + spec: { + containers: [ + { + name: 'kibana', + image: 'docker.elastic.co/kibana/kibana:8.12.0', + ports: [{ containerPort: 5601 }], + env: [ + { name: 'ELASTICSEARCH_HOSTS', value: `http://${this.ES_NAME}:9200` }, + { name: 'ELASTICSEARCH_USERNAME', value: 'elastic' }, + { + name: 'ELASTICSEARCH_PASSWORD', + valueFrom: { + secretKeyRef: { name: 'elasticsearch-credentials', key: 'ELASTIC_PASSWORD' }, + }, + }, + { name: 'SERVER_NAME', value: 'kibana' }, + { name: 'XPACK_SECURITY_ENABLED', value: 'true' }, + ], + resources: { + requests: { cpu: '200m', memory: '512Mi' }, + limits: { cpu: '1000m', memory: '1Gi' }, + }, + readinessProbe: { + httpGet: { path: '/api/status', port: 5601 as any }, + initialDelaySeconds: 30, + periodSeconds: 10, + }, + livenessProbe: { + httpGet: { path: '/api/status', port: 5601 as any }, + initialDelaySeconds: 60, + periodSeconds: 30, + }, + }, + ], + }, + }, + }, + }; + + try { + await appsApi.replaceNamespacedDeployment(this.KIBANA_NAME, this.ES_NAMESPACE, deployment); + } catch { + await appsApi.createNamespacedDeployment(this.ES_NAMESPACE, deployment); + } + + const service: k8s.V1Service = { + apiVersion: 'v1', + kind: 'Service', + metadata: { name: this.KIBANA_NAME, namespace: this.ES_NAMESPACE }, + spec: { + selector: { app: this.KIBANA_NAME }, + ports: [{ port: 5601, targetPort: 5601 as any }], + type: 'ClusterIP', + }, + }; + + try { + await coreApi.replaceNamespacedService(this.KIBANA_NAME, this.ES_NAMESPACE, service); + } catch { + await coreApi.createNamespacedService(this.ES_NAMESPACE, service); + } + + this.logger.log('Kibana deployed'); + } + + /** + * Get Elasticsearch connection info for apps + */ + getConnectionInfo(): { host: string; port: number; username: string; password: string } { + return { + host: `${this.ES_NAME}.${this.ES_NAMESPACE}.svc.cluster.local`, + 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: `user-${userId.split('-')[0]}`, + 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 { + const userPrefix = userId.split('-')[0]; + return `logs-user-${userPrefix}-*`; + } + + /** + * Build Elasticsearch query to filter logs by user/owner + * Users can only see logs from their own applications + */ + getUserLogsQuery(userId: string): { query: { bool: { must: any } } } { + return { + query: { + bool: { + must: { + term: { + 'kubernetes.labels.owner': userId, + }, + }, + }, + }, + }; + } +}