feat: api to pull service + k8s attrs linkings

.changeset/serious-experts-whisper.md

@@ -0,0 +1,6 @@
+---
+'@hyperdx/api': patch
+'@hyperdx/app': patch
+---
+
+feat: api to pull service + k8s attrs linkings

packages/api/src/clickhouse/__tests__/clickhouse.test.ts

@@ -81,7 +81,7 @@ Array [
   Object {
     "_host": "",
     "_platform": "nodejs",
-    "_service": "",
+    "_service": "test-service",
     "body": "",
     "duration": -1641340800001,
     "severity_text": "",
@@ -92,7 +92,7 @@ Array [
   Object {
     "_host": "",
     "_platform": "nodejs",
-    "_service": "",
+    "_service": "test-service",
     "body": "",
     "duration": -1641340800000,
     "severity_text": "",

packages/api/src/fixtures.ts

@@ -161,6 +161,7 @@ export function buildEvent({
   type = LogType.Log,
   end_timestamp = 0,
   span_name,
+  service = 'test-service',
   ...properties
 }: {
   level?: string;
@@ -170,6 +171,7 @@ export function buildEvent({
   type?: LogType;
   end_timestamp?: number; //ms timestamp
   span_name?: string;
+  service?: string;
 } & {
   [key: string]: number | string | boolean;
 }): LogStreamModel {
@@ -203,6 +205,7 @@ export function buildEvent({
     observed_timestamp: `${ts}000000`,
     _source: source,
     _platform: platform,
+    _service: service,
     severity_text: level,
     // @ts-ignore
     end_timestamp: `${end_timestamp}000000`,

packages/api/src/routers/api/__tests__/chart.test.ts

@@ -0,0 +1,125 @@
+import ms from 'ms';
+
+import * as clickhouse from '@/clickhouse';
+import {
+  buildEvent,
+  clearClickhouseTables,
+  clearDBCollections,
+  clearRedis,
+  closeDB,
+  getLoggedInAgent,
+  getServer,
+} from '@/fixtures';
+
+describe('charts router', () => {
+  const server = getServer();
+
+  beforeAll(async () => {
+    await server.start();
+  });
+
+  afterEach(async () => {
+    await clearDBCollections();
+    await clearClickhouseTables();
+    await clearRedis();
+  });
+
+  afterAll(async () => {
+    await server.closeHttpServer();
+    await closeDB();
+  });
+
+  it('GET /chart/services', async () => {
+    const now = Date.now();
+    const { agent, team } = await getLoggedInAgent(server);
+
+    await clickhouse.bulkInsertTeamLogStream(
+      team.logStreamTableVersion,
+      team.id,
+      [
+        buildEvent({
+          timestamp: now,
+          service: 'service1',
+          'k8s.namespace.name': 'namespace1',
+          'k8s.pod.name': 'pod1',
+          'k8s.pod.uid': 'uid1',
+        }),
+        buildEvent({
+          timestamp: now,
+          service: 'service1',
+          'k8s.namespace.name': 'namespace1',
+          'k8s.pod.name': 'pod2',
+          'k8s.pod.uid': 'uid2',
+        }),
+        buildEvent({
+          timestamp: now - ms('1d'),
+          service: 'service2',
+          'k8s.namespace.name': 'namespace2',
+          'k8s.pod.name': 'pod3',
+          'k8s.pod.uid': 'uid3',
+        }),
+      ],
+    );
+
+    const results = await agent.get('/chart/services').expect(200);
+    expect(results.body.data).toMatchInlineSnapshot(`
+Object {
+  "service1": Array [
+    Object {
+      "k8s.namespace.name": "namespace1",
+      "k8s.pod.name": "pod1",
+      "k8s.pod.uid": "uid1",
+    },
+    Object {
+      "k8s.namespace.name": "namespace1",
+      "k8s.pod.name": "pod2",
+      "k8s.pod.uid": "uid2",
+    },
+  ],
+  "service2": Array [
+    Object {
+      "k8s.namespace.name": "namespace2",
+      "k8s.pod.name": "pod3",
+      "k8s.pod.uid": "uid3",
+    },
+  ],
+}
+`);
+  });
+
+  it('GET /chart/services (missing custom attributes)', async () => {
+    const now = Date.now();
+    const { agent, team } = await getLoggedInAgent(server);
+
+    await clickhouse.bulkInsertTeamLogStream(
+      team.logStreamTableVersion,
+      team.id,
+      [
+        buildEvent({
+          timestamp: now,
+          service: 'service1',
+        }),
+        buildEvent({
+          timestamp: now,
+          service: 'service1',
+        }),
+        buildEvent({
+          timestamp: now - ms('1d'),
+          service: 'service2',
+        }),
+      ],
+    );
+
+    const results = await agent.get('/chart/services').expect(200);
+    expect(results.body.data).toMatchInlineSnapshot(`
+Object {
+  "service1": Array [
+    Object {},
+  ],
+  "service2": Array [
+    Object {},
+  ],
+}
+`);
+  });
+});

packages/api/src/routers/api/chart.ts

@@ -1,16 +1,101 @@
 import opentelemetry, { SpanStatusCode } from '@opentelemetry/api';
 import express from 'express';
-import { isNumber, parseInt } from 'lodash';
+import { isNumber } from 'lodash';
+import ms from 'ms';
 import { z } from 'zod';
 import { validateRequest } from 'zod-express-middleware';
 
 import * as clickhouse from '@/clickhouse';
+import { buildSearchColumnName } from '@/clickhouse/searchQueryParser';
 import { getTeam } from '@/controllers/team';
-import logger from '@/utils/logger';
+import { SimpleCache } from '@/utils/redis';
 import { chartSeriesSchema } from '@/utils/zod';
 
 const router = express.Router();
 
+router.get('/services', async (req, res, next) => {
+  try {
+    const teamId = req.user?.team;
+    if (teamId == null) {
+      return res.sendStatus(403);
+    }
+    const team = await getTeam(teamId);
+    if (team == null) {
+      return res.sendStatus(403);
+    }
+
+    const FIELDS = ['k8s.namespace.name', 'k8s.pod.name', 'k8s.pod.uid'];
+    const nowInMs = Date.now();
+    const startTime = nowInMs - ms('5d');
+    const endTime = nowInMs;
+
+    const propertyTypeMappingsModel =
+      await clickhouse.buildLogsPropertyTypeMappingsModel(
+        team.logStreamTableVersion,
+        teamId.toString(),
+        startTime,
+        endTime,
+      );
+
+    const targetGroupByFields: string[] = ['service'];
+    // make sure all custom fields exist
+    for (const f of FIELDS) {
+      if (buildSearchColumnName(propertyTypeMappingsModel.get(f), f)) {
+        targetGroupByFields.push(f);
+      }
+    }
+
+    const MAX_NUM_GROUPS = 2000;
+
+    const simpleCache = new SimpleCache<
+      Awaited<ReturnType<typeof clickhouse.getMultiSeriesChart>>
+    >(`chart-services-${teamId}`, ms('10m'), () =>
+      clickhouse.getMultiSeriesChart({
+        series: [
+          {
+            aggFn: clickhouse.AggFn.Count,
+            groupBy: targetGroupByFields,
+            table: 'logs',
+            type: 'table',
+            where: '',
+          },
+        ],
+        endTime,
+        granularity: undefined,
+        maxNumGroups: MAX_NUM_GROUPS,
+        startTime,
+        tableVersion: team.logStreamTableVersion,
+        teamId: teamId.toString(),
+        seriesReturnType: clickhouse.SeriesReturnType.Column,
+      }),
+    );
+
+    const results = await simpleCache.get();
+    // restructure service maps
+    const serviceMap: Record<string, Record<string, string>[]> = {};
+    for (const row of results.data) {
+      const values = row.group;
+      const service = values[0];
+      if (!(service in serviceMap)) {
+        serviceMap[service] = [];
+      }
+      const k8sAttrs: Record<string, string> = {};
+      for (let i = 1; i < values.length; i++) {
+        const field = targetGroupByFields[i];
+        const value = values[i];
+        k8sAttrs[field] = value;
+      }
+      serviceMap[service].push(k8sAttrs);
+    }
+
+    res.json({
+      data: serviceMap,
+    });
+  } catch (e) {
+    next(e);
+  }
+});
+
 router.post(
   '/series',
   validateRequest({