Add unified logs platform with Helm-managed central Elasticsearch.

Deploy cloudhost-logging on cluster registration, ship app and optional service logs to ES with owner isolation, and fix Kibana 8.12 auth via kibana_system.

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

backend/src/kubernetes/kubernetes.service.ts

@@ -41,6 +41,8 @@ interface ManifestContext {
   enableElasticsearch: boolean;
   elasticsearchVersion: string;
   logPaths: string[];
+  ownerId: string;
+  applicationId: string;
 }
 
 type StorageUsageSlice = {
@@ -183,6 +185,8 @@ export class KubernetesService implements OnModuleInit {
       elasticsearch: {
         enabled: app.enableElasticsearch || false,
         logPaths: app.logPaths || [],
+        ownerId: app.userId,
+        applicationId: app.id,
       },
       changeCause: `Deploy ${imageUri} at ${new Date().toISOString()}`,
     };
@@ -247,6 +251,8 @@ export class KubernetesService implements OnModuleInit {
         enableElasticsearch: app.enableElasticsearch || false,
         elasticsearchVersion: app.elasticsearchVersion || '8.12',
         logPaths: app.logPaths || [],
+        ownerId: app.userId,
+        applicationId: app.id,
       };
       await this.applyIngress(networkingApi, ctx, customDomain);
       this.logger.log(`Updated ingress for ${app.name} via K8s API (customDomain: ${customDomain || 'none'})`);
@@ -305,6 +311,8 @@ export class KubernetesService implements OnModuleInit {
       enableElasticsearch: app.enableElasticsearch || false,
       elasticsearchVersion: app.elasticsearchVersion || '8.12',
       logPaths: app.logPaths || [],
+      ownerId: app.userId,
+      applicationId: app.id,
     };
 
     const manifests: Record<string, any> = {};
@@ -689,7 +697,15 @@ export class KubernetesService implements OnModuleInit {
   /**
    * Build Fluent Bit configuration for log collection
    */
-  private buildFluentBitConfig(appName: string, namespace: string, runtime: string, customLogPaths?: string[]): string {
+  private buildFluentBitConfig(
+    appName: string,
+    namespace: string,
+    runtime: string,
+    ownerId: string,
+    applicationId: string,
+    workload: string,
+    customLogPaths?: string[],
+  ): string {
     const logPaths = customLogPaths && customLogPaths.length > 0 
       ? customLogPaths 
       : this.getDefaultLogPaths(runtime);
@@ -714,8 +730,12 @@ export class KubernetesService implements OnModuleInit {
     Name          record_modifier
     Match         *
     Record        app ${appName}
+    Record        applicationName ${appName}
     Record        namespace ${namespace}
     Record        runtime ${runtime}
+    Record        ownerId ${ownerId}
+    Record        applicationId ${applicationId}
+    Record        workload ${workload}
 
 [FILTER]
     Name          parser
@@ -763,7 +783,15 @@ export class KubernetesService implements OnModuleInit {
         labels: { app: ctx.appName },
       },
       data: {
-        'fluent-bit.conf': this.buildFluentBitConfig(ctx.appName, ctx.namespace, ctx.runtime, customLogPaths),
+        'fluent-bit.conf': this.buildFluentBitConfig(
+          ctx.appName,
+          ctx.namespace,
+          ctx.runtime,
+          ctx.ownerId,
+          ctx.applicationId,
+          'app',
+          customLogPaths,
+        ),
         'parsers.conf': `
 [PARSER]
     Name        json
@@ -788,6 +816,127 @@ export class KubernetesService implements OnModuleInit {
     this.logger.log(`Created Fluent Bit ConfigMap for ${ctx.appName}`);
   }
 
+  private buildWorkloadFluentBitConfig(
+    ctx: ManifestContext,
+    workload: 'redis' | 'rabbitmq' | 'database',
+    resourceName: string,
+  ): string {
+    const logGlob = `/var/log/pods/*${resourceName}*/*/*.log`;
+    return `
+[SERVICE]
+    Flush         5
+    Daemon        Off
+    Log_Level     info
+    Parsers_File  /fluent-bit/etc/parsers.conf
+
+[INPUT]
+    Name              tail
+    Path              ${logGlob}
+    Tag               ${workload}.${resourceName}
+    Refresh_Interval  5
+    Mem_Buf_Limit     5MB
+    Skip_Long_Lines   On
+    Parser            docker
+
+[FILTER]
+    Name          record_modifier
+    Match         *
+    Record        app ${ctx.appName}
+    Record        applicationName ${ctx.appName}
+    Record        namespace ${ctx.namespace}
+    Record        ownerId ${ctx.ownerId}
+    Record        applicationId ${ctx.applicationId}
+    Record        workload ${workload}
+
+[OUTPUT]
+    Name          es
+    Match         *
+    Host          \${ES_HOST}
+    Port          \${ES_PORT}
+    HTTP_User     elastic
+    HTTP_Passwd   \${ES_PASSWORD}
+    Index         logs-${ctx.namespace}-${ctx.appName}
+    Logstash_Format On
+    Logstash_Prefix logs-${ctx.namespace}
+    Suppress_Type_Name On
+    tls           Off
+    Retry_Limit   3
+`;
+  }
+
+  private async attachWorkloadLogShipper(
+    coreApi: k8s.CoreV1Api,
+    ctx: ManifestContext,
+    workload: 'redis' | 'rabbitmq' | 'database',
+    resourceName: string,
+  ): Promise<{ containers: k8s.V1Container[]; volumes: k8s.V1Volume[] }> {
+    if (!ctx.enableElasticsearch) {
+      return { containers: [], volumes: [] };
+    }
+
+    const configMapName = `${resourceName}-log-shipper-config`;
+    const configMap = {
+      apiVersion: 'v1',
+      kind: 'ConfigMap',
+      metadata: {
+        name: configMapName,
+        namespace: ctx.namespace,
+        labels: { app: resourceName, 'cloudhost.io/log-shipper': 'true' },
+      },
+      data: {
+        'fluent-bit.conf': this.buildWorkloadFluentBitConfig(ctx, workload, resourceName),
+        'parsers.conf': `
+[PARSER]
+    Name        docker
+    Format      json
+    Time_Key    time
+    Time_Format %Y-%m-%dT%H:%M:%S.%L
+`,
+      },
+    };
+
+    try {
+      await coreApi.replaceNamespacedConfigMap(configMapName, ctx.namespace, configMap);
+    } catch {
+      await coreApi.createNamespacedConfigMap(ctx.namespace, configMap);
+    }
+
+    return {
+      containers: [
+        {
+          name: 'log-shipper',
+          image: 'fluent/fluent-bit:2.2',
+          resources: {
+            requests: { cpu: '10m', memory: '32Mi' },
+            limits: { cpu: '50m', memory: '64Mi' },
+          },
+          volumeMounts: [
+            { name: 'varlogpods', mountPath: '/var/log/pods', readOnly: true },
+            { name: 'log-shipper-config', mountPath: '/fluent-bit/etc' },
+          ],
+          env: [
+            { name: 'ES_HOST', value: 'elasticsearch.logging.svc.cluster.local' },
+            { name: 'ES_PORT', value: '9200' },
+            {
+              name: 'ES_PASSWORD',
+              valueFrom: {
+                secretKeyRef: {
+                  name: 'elasticsearch-credentials',
+                  key: 'ELASTIC_PASSWORD',
+                  optional: true,
+                },
+              },
+            },
+          ],
+        },
+      ],
+      volumes: [
+        { name: 'varlogpods', hostPath: { path: '/var/log/pods', type: 'Directory' } },
+        { name: 'log-shipper-config', configMap: { name: configMapName } },
+      ],
+    };
+  }
+
   private async applyService(coreApi: k8s.CoreV1Api, ctx: ManifestContext): Promise<any> {
     const service: k8s.V1Service = {
       apiVersion: 'v1',
@@ -953,6 +1102,8 @@ export class KubernetesService implements OnModuleInit {
         throw new Error(`Unsupported database type: ${dbType}`);
     }
 
+    const dbLogShipper = await this.attachWorkloadLogShipper(coreApi, ctx, 'database', dbName);
+
     const dbDeployment: k8s.V1Deployment = {
       apiVersion: 'apps/v1',
       kind: 'Deployment',
@@ -977,9 +1128,11 @@ export class KubernetesService implements OnModuleInit {
                 readinessProbe,
                 livenessProbe,
               },
+              ...dbLogShipper.containers,
             ],
             volumes: [
               { name: 'db-storage', persistentVolumeClaim: { claimName: dbName } },
+              ...dbLogShipper.volumes,
             ],
           },
         },
@@ -1091,6 +1244,8 @@ export class KubernetesService implements OnModuleInit {
       await coreApi.createNamespacedSecret(ctx.namespace, redisSecret);
     }
 
+    const logShipper = await this.attachWorkloadLogShipper(coreApi, ctx, 'redis', redisName);
+
     // Create Redis Deployment
     const redisDeployment: k8s.V1Deployment = {
       apiVersion: 'apps/v1',
@@ -1134,12 +1289,14 @@ export class KubernetesService implements OnModuleInit {
                   periodSeconds: 20,
                 },
               },
+              ...logShipper.containers,
             ],
             volumes: [
               {
                 name: 'redis-data',
                 persistentVolumeClaim: { claimName: `${redisName}-data` },
               },
+              ...logShipper.volumes,
             ],
           },
         },
@@ -1205,6 +1362,8 @@ export class KubernetesService implements OnModuleInit {
       await coreApi.createNamespacedSecret(ctx.namespace, rabbitSecret);
     }
 
+    const rabbitLogShipper = await this.attachWorkloadLogShipper(coreApi, ctx, 'rabbitmq', rabbitName);
+
     // Create RabbitMQ Deployment
     const rabbitDeployment: k8s.V1Deployment = {
       apiVersion: 'apps/v1',
@@ -1258,12 +1417,14 @@ export class KubernetesService implements OnModuleInit {
                   timeoutSeconds: 10,
                 },
               },
+              ...rabbitLogShipper.containers,
             ],
             volumes: [
               {
                 name: 'rabbitmq-data',
                 persistentVolumeClaim: { claimName: `${rabbitName}-data` },
               },
+              ...rabbitLogShipper.volumes,
             ],
           },
         },