feat(logging): add Elasticsearch admin and user logs API

Add admin endpoints for stack deploy/status and user-facing logs
query API. Register controllers and service in KubernetesModule.

Co-authored-by: Cursor <cursoragent@cursor.com>

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
+`,
+    };
+  }
+}

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],
+  controllers: [ElasticsearchController, LogsController],
-  exports: [KubernetesService, HelmService],
+  providers: [KubernetesService, HelmService, ElasticsearchService],
+  exports: [KubernetesService, HelmService, ElasticsearchService],
 })
 export class KubernetesModule {}

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<string, string> = {
+      '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<string, string> };
+    }> = [
+      {
+        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',
+      },
+    };
+  }
+}