diff --git a/backend/src/kubernetes/elasticsearch.controller.ts b/backend/src/kubernetes/elasticsearch.controller.ts new file mode 100644 index 0000000..02e0817 --- /dev/null +++ b/backend/src/kubernetes/elasticsearch.controller.ts @@ -0,0 +1,127 @@ +import { Controller, Post, Get, Delete, Query, UseGuards } from '@nestjs/common'; +import { ApiTags, ApiOperation, ApiBearerAuth, ApiQuery, ApiResponse } from '@nestjs/swagger'; +import { AuthGuard } from '@nestjs/passport'; +import { RolesGuard } from '../common/guards/roles.guard'; +import { Roles } from '../common/decorators/roles.decorator'; +import { UserRole } from '../common/enums'; +import { ElasticsearchService } from './elasticsearch.service'; + +@ApiTags('Admin - Elasticsearch') +@ApiBearerAuth() +@Controller('admin/elasticsearch') +@UseGuards(AuthGuard('jwt'), RolesGuard) +@Roles(UserRole.ADMIN, UserRole.TECHNICAL) +export class ElasticsearchController { + constructor(private readonly esService: ElasticsearchService) {} + + @Get('status') + @ApiOperation({ summary: 'Check if central Elasticsearch is deployed and get health' }) + @ApiQuery({ name: 'clusterId', required: false, description: 'Target cluster ID' }) + @ApiResponse({ status: 200, description: 'Elasticsearch status' }) + async getStatus(@Query('clusterId') clusterId?: string) { + const isDeployed = await this.esService.isDeployed(clusterId); + const health = isDeployed ? await this.esService.getHealth(clusterId) : null; + const connectionInfo = this.esService.getConnectionInfo(); + + return { + deployed: isDeployed, + health, + namespace: 'logging', + elasticsearch: { + host: connectionInfo.host, + port: connectionInfo.port, + internalUrl: `http://${connectionInfo.host}:${connectionInfo.port}`, + }, + kibana: { + host: 'kibana.logging.svc.cluster.local', + port: 5601, + internalUrl: 'http://kibana.logging.svc.cluster.local:5601', + }, + }; + } + + @Post('deploy') + @ApiOperation({ summary: 'Deploy central Elasticsearch + Kibana stack' }) + @ApiQuery({ name: 'clusterId', required: false, description: 'Target cluster ID' }) + @ApiResponse({ status: 201, description: 'Elasticsearch deployed successfully' }) + async deploy(@Query('clusterId') clusterId?: string) { + const alreadyDeployed = await this.esService.isDeployed(clusterId); + + if (alreadyDeployed) { + const health = await this.esService.getHealth(clusterId); + return { + success: true, + message: 'Elasticsearch is already deployed', + status: 'existing', + health, + }; + } + + const result = await this.esService.deploy(clusterId); + + return { + success: true, + message: 'Elasticsearch and Kibana deployed successfully. Please wait 2-3 minutes for pods to be ready.', + status: 'created', + credentials: { + username: 'elastic', + password: result.esPassword, + note: 'Store this password securely. It is required for Kibana and API access.', + }, + endpoints: { + elasticsearch: 'http://elasticsearch.logging.svc.cluster.local:9200', + kibana: result.kibanaUrl, + }, + fluentBitConfig: { + host: 'elasticsearch.logging.svc.cluster.local', + port: 9200, + username: 'elastic', + password: result.esPassword, + }, + }; + } + + @Delete('undeploy') + @ApiOperation({ summary: 'Remove central Elasticsearch stack (preserves data)' }) + @ApiQuery({ name: 'clusterId', required: false, description: 'Target cluster ID' }) + @ApiResponse({ status: 200, description: 'Elasticsearch undeployed' }) + async undeploy(@Query('clusterId') clusterId?: string) { + await this.esService.undeploy(clusterId); + + return { + success: true, + message: 'Elasticsearch stack removed. PVC with data is preserved.', + note: 'To permanently delete data, manually delete the PVC: kubectl delete pvc elasticsearch-data -n logging', + }; + } + + @Get('credentials') + @ApiOperation({ summary: 'Get Elasticsearch credentials for Fluent Bit' }) + @ApiResponse({ status: 200, description: 'Fluent Bit credentials' }) + async getCredentials() { + const creds = this.esService.getFluentBitCredentials(); + const connInfo = this.esService.getConnectionInfo(); + + return { + elasticsearch: { + host: connInfo.host, + port: connInfo.port, + username: creds.username, + password: creds.password, + }, + fluentBitConfig: ` +[OUTPUT] + Name es + Match * + Host ${connInfo.host} + Port ${connInfo.port} + HTTP_User ${creds.username} + HTTP_Passwd ${creds.password} + Logstash_Format On + Logstash_Prefix logs + Suppress_Type_Name On + tls Off +`, + }; + } +} diff --git a/backend/src/kubernetes/kubernetes.module.ts b/backend/src/kubernetes/kubernetes.module.ts index f9fd805..aa574b0 100644 --- a/backend/src/kubernetes/kubernetes.module.ts +++ b/backend/src/kubernetes/kubernetes.module.ts @@ -1,11 +1,15 @@ import { Module, forwardRef } from '@nestjs/common'; import { KubernetesService } from './kubernetes.service'; import { HelmService } from './helm.service'; +import { ElasticsearchService } from './elasticsearch.service'; +import { ElasticsearchController } from './elasticsearch.controller'; +import { LogsController } from './logs.controller'; import { ClustersModule } from '../clusters/clusters.module'; @Module({ imports: [forwardRef(() => ClustersModule)], - providers: [KubernetesService, HelmService], - exports: [KubernetesService, HelmService], + controllers: [ElasticsearchController, LogsController], + providers: [KubernetesService, HelmService, ElasticsearchService], + exports: [KubernetesService, HelmService, ElasticsearchService], }) export class KubernetesModule {} diff --git a/backend/src/kubernetes/logs.controller.ts b/backend/src/kubernetes/logs.controller.ts new file mode 100644 index 0000000..625541d --- /dev/null +++ b/backend/src/kubernetes/logs.controller.ts @@ -0,0 +1,403 @@ +import { + Controller, + Get, + Query, + UseGuards, + Request, + BadRequestException, + NotFoundException, +} from '@nestjs/common'; +import { + ApiTags, + ApiOperation, + ApiBearerAuth, + ApiQuery, + ApiResponse, +} from '@nestjs/swagger'; +import { AuthGuard } from '@nestjs/passport'; +import { ElasticsearchService } from './elasticsearch.service'; + +interface AuthenticatedRequest { + user: { + sub: string; + email: string; + }; +} + +@ApiTags('Logs') +@ApiBearerAuth() +@Controller('logs') +@UseGuards(AuthGuard('jwt')) +export class LogsController { + constructor(private readonly esService: ElasticsearchService) {} + + @Get() + @ApiOperation({ summary: 'Get logs for authenticated user\'s applications' }) + @ApiQuery({ name: 'appId', required: false, description: 'Filter by application ID' }) + @ApiQuery({ name: 'level', required: false, description: 'Filter by log level (error, warn, info, debug)' }) + @ApiQuery({ name: 'from', required: false, description: 'Start time (ISO 8601 format)' }) + @ApiQuery({ name: 'to', required: false, description: 'End time (ISO 8601 format)' }) + @ApiQuery({ name: 'search', required: false, description: 'Full-text search in log messages' }) + @ApiQuery({ name: 'page', required: false, description: 'Page number (default: 1)' }) + @ApiQuery({ name: 'limit', required: false, description: 'Results per page (default: 100, max: 1000)' }) + @ApiResponse({ status: 200, description: 'User logs' }) + @ApiResponse({ status: 400, description: 'Invalid query parameters' }) + async getUserLogs( + @Request() req: AuthenticatedRequest, + @Query('appId') appId?: string, + @Query('level') level?: string, + @Query('from') from?: string, + @Query('to') to?: string, + @Query('search') search?: string, + @Query('page') page?: string, + @Query('limit') limit?: string, + ) { + const userId = req.user.sub; + const pageNum = parseInt(page || '1', 10); + const limitNum = Math.min(parseInt(limit || '100', 10), 1000); + const offset = (pageNum - 1) * limitNum; + + // Validate log level + if (level && !['error', 'warn', 'info', 'debug', 'trace'].includes(level.toLowerCase())) { + throw new BadRequestException('Invalid log level. Use: error, warn, info, debug, or trace'); + } + + // Validate date formats + if (from && isNaN(Date.parse(from))) { + throw new BadRequestException('Invalid "from" date format. Use ISO 8601 format.'); + } + if (to && isNaN(Date.parse(to))) { + throw new BadRequestException('Invalid "to" date format. Use ISO 8601 format.'); + } + + // Build Elasticsearch query + const baseQuery = this.esService.getUserLogsQuery(userId); + const must: any[] = [baseQuery.query.bool.must]; + + // Add application filter + if (appId) { + must.push({ + term: { 'kubernetes.labels.app': appId }, + }); + } + + // Add level filter + if (level) { + must.push({ + term: { level: level.toLowerCase() }, + }); + } + + // Add time range filter + if (from || to) { + const rangeFilter: any = { range: { '@timestamp': {} } }; + if (from) rangeFilter.range['@timestamp'].gte = from; + if (to) rangeFilter.range['@timestamp'].lte = to; + must.push(rangeFilter); + } + + // Add full-text search + if (search) { + must.push({ + multi_match: { + query: search, + fields: ['message', 'log', 'msg', 'error.message'], + type: 'phrase_prefix', + }, + }); + } + + const query = { + query: { + bool: { + must, + }, + }, + sort: [{ '@timestamp': 'desc' }], + from: offset, + size: limitNum, + }; + + return { + query, + meta: { + page: pageNum, + limit: limitNum, + userId, + filters: { + appId: appId || null, + level: level || null, + from: from || null, + to: to || null, + search: search || null, + }, + }, + usage: { + description: 'Execute this query against Elasticsearch to get logs', + endpoint: 'POST /logs-*/_search', + note: 'Use the Elasticsearch endpoint provided by admin to execute queries', + }, + }; + } + + @Get('stream') + @ApiOperation({ summary: 'Get live log stream query for user\'s applications' }) + @ApiQuery({ name: 'appId', required: false, description: 'Filter by application ID' }) + @ApiResponse({ status: 200, description: 'Stream query configuration' }) + async getStreamConfig( + @Request() req: AuthenticatedRequest, + @Query('appId') appId?: string, + ) { + const userId = req.user.sub; + + // Build query for streaming + const baseQuery = this.esService.getUserLogsQuery(userId); + const must: any[] = [baseQuery.query.bool.must]; + + if (appId) { + must.push({ + term: { 'kubernetes.labels.app': appId }, + }); + } + + // Add time filter for last 5 minutes + must.push({ + range: { + '@timestamp': { + gte: 'now-5m', + }, + }, + }); + + const query = { + query: { + bool: { + must, + }, + }, + sort: [{ '@timestamp': 'asc' }], + size: 100, + }; + + return { + query, + meta: { + userId, + appId: appId || 'all', + refreshInterval: '5s', + }, + usage: { + description: 'Poll this query every 5 seconds to get new logs', + note: 'Use search_after for efficient pagination in streaming mode', + }, + }; + } + + @Get('stats') + @ApiOperation({ summary: 'Get log statistics for user\'s applications' }) + @ApiQuery({ name: 'appId', required: false, description: 'Filter by application ID' }) + @ApiQuery({ name: 'period', required: false, description: 'Time period: 1h, 6h, 24h, 7d (default: 24h)' }) + @ApiResponse({ status: 200, description: 'Log statistics' }) + async getLogStats( + @Request() req: AuthenticatedRequest, + @Query('appId') appId?: string, + @Query('period') period?: string, + ) { + const userId = req.user.sub; + + // Convert period to time range + const periodMap: Record = { + '1h': 'now-1h', + '6h': 'now-6h', + '24h': 'now-24h', + '7d': 'now-7d', + }; + const timeRange = periodMap[period || '24h'] || 'now-24h'; + + // Build aggregation query + const baseQuery = this.esService.getUserLogsQuery(userId); + const must: any[] = [baseQuery.query.bool.must]; + + if (appId) { + must.push({ + term: { 'kubernetes.labels.app': appId }, + }); + } + + must.push({ + range: { + '@timestamp': { + gte: timeRange, + }, + }, + }); + + const aggregationQuery = { + query: { + bool: { + must, + }, + }, + size: 0, + aggs: { + by_level: { + terms: { + field: 'level', + size: 10, + }, + }, + by_app: { + terms: { + field: 'kubernetes.labels.app', + size: 50, + }, + }, + over_time: { + date_histogram: { + field: '@timestamp', + fixed_interval: period === '1h' ? '5m' : period === '6h' ? '30m' : '1h', + }, + aggs: { + by_level: { + terms: { + field: 'level', + size: 5, + }, + }, + }, + }, + error_count: { + filter: { + term: { level: 'error' }, + }, + }, + warn_count: { + filter: { + term: { level: 'warn' }, + }, + }, + }, + }; + + return { + query: aggregationQuery, + meta: { + userId, + appId: appId || 'all', + period: period || '24h', + timeRange, + }, + usage: { + description: 'Execute this aggregation query to get log statistics', + endpoint: 'POST /logs-*/_search', + }, + }; + } + + @Get('errors') + @ApiOperation({ summary: 'Get recent errors for user\'s applications' }) + @ApiQuery({ name: 'appId', required: false, description: 'Filter by application ID' }) + @ApiQuery({ name: 'hours', required: false, description: 'Hours to look back (default: 24)' }) + @ApiQuery({ name: 'limit', required: false, description: 'Max errors to return (default: 50)' }) + @ApiResponse({ status: 200, description: 'Recent errors' }) + async getRecentErrors( + @Request() req: AuthenticatedRequest, + @Query('appId') appId?: string, + @Query('hours') hours?: string, + @Query('limit') limit?: string, + ) { + const userId = req.user.sub; + const hoursNum = parseInt(hours || '24', 10); + const limitNum = Math.min(parseInt(limit || '50', 10), 500); + + // Build error query + const baseQuery = this.esService.getUserLogsQuery(userId); + const must: any[] = [baseQuery.query.bool.must]; + + if (appId) { + must.push({ + term: { 'kubernetes.labels.app': appId }, + }); + } + + must.push({ + term: { level: 'error' }, + }); + + must.push({ + range: { + '@timestamp': { + gte: `now-${hoursNum}h`, + }, + }, + }); + + const query = { + query: { + bool: { + must, + }, + }, + sort: [{ '@timestamp': 'desc' }], + size: limitNum, + _source: ['@timestamp', 'message', 'log', 'error', 'kubernetes.labels.app', 'kubernetes.pod_name'], + }; + + return { + query, + meta: { + userId, + appId: appId || 'all', + lookbackHours: hoursNum, + limit: limitNum, + }, + usage: { + description: 'Execute this query to get recent errors', + endpoint: 'POST /logs-*/_search', + }, + }; + } + + @Get('kibana-url') + @ApiOperation({ summary: 'Get Kibana URL for user\'s application logs' }) + @ApiQuery({ name: 'appId', required: false, description: 'Application ID to filter' }) + @ApiResponse({ status: 200, description: 'Kibana discovery URL' }) + async getKibanaUrl( + @Request() req: AuthenticatedRequest, + @Query('appId') appId?: string, + ) { + const userId = req.user.sub; + const connInfo = this.esService.getConnectionInfo(); + + const filters: Array<{ + meta: { key: string; negate: boolean }; + query: { match_phrase: Record }; + }> = [ + { + meta: { key: 'kubernetes.labels.owner', negate: false }, + query: { match_phrase: { 'kubernetes.labels.owner': userId } }, + }, + ]; + + if (appId) { + filters.push({ + meta: { key: 'kubernetes.labels.app', negate: false }, + query: { match_phrase: { 'kubernetes.labels.app': appId } }, + }); + } + + const rison = encodeURIComponent(JSON.stringify(filters)); + + return { + kibana: { + baseUrl: `http://${connInfo.host.replace('elasticsearch', 'kibana')}:5601`, + discoverUrl: `/app/discover#/?_g=(time:(from:now-24h,to:now))&_a=(filters:!${rison})`, + note: 'Access Kibana through your cluster ingress or port-forward', + }, + portForward: { + command: 'kubectl port-forward svc/kibana 5601:5601 -n logging', + localUrl: 'http://localhost:5601', + }, + }; + } +}