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 <cursoragent@cursor.com>
This commit is contained in:
keyhan
@@ -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<boolean> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
// 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<void> {
|
||||
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,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user